JQC_multi_tenant/tests/test_issue_links.py

"""
tests/test_issue_links.py
-------------------------
phase57 — link a duplicate to its original, or two issues about the same thing.

Two things are being pinned here.

**The direction convention.** One row is stored per pair and shown on BOTH
issues, so the same row has to read differently at each end ("Duplicate of #B"
on one, "Duplicated by #A" on the other). That is what makes `exists_between()`
necessary: `(A,B)` and `(B,A)` are distinct rows to the database but the same
link to a person, and the UniqueConstraint only covers the stored direction.

**Scope, which is the part that actually matters.** A link is a pointer that
exposes the far issue's id, description, facility and status. Three surfaces
have to hold the line and only one of them is a real boundary:

    _readable_links()  filters what the panel RENDERS
    link_search()      scopes what the picker FINDS
    add_link()         re-checks on POST — the search is a convenience

If any one of them is unfiltered, a customer reads an issue at a facility they
hold no assignment to, simply because one of our staff linked it.

Cross-TENANT isolation is not tested here and cannot be: `issue_links` lives in
the tenant database and every query routes through RoutingSession, so an id
from another tenant does not resolve at all. See tests/test_tenant_isolation.py
for that layer. What follows is scope WITHIN one tenant.
"""

import pytest


@pytest.fixture
def client(app):
    with app.app_context():
        from app import db
        db.drop_all()
        db.create_all()
        yield app.test_client()
        db.session.remove()


# ── Builders ─────────────────────────────────────────────────────────────────

_n = {'i': 0}


def _uniq():
    _n['i'] += 1
    return _n['i']


def _user(role='admin', **kw):
    from app import db
    from app.models.user import User
    n = _uniq()
    u = User(username=kw.pop('username', f'{role}{n}'),
             full_name=f'{role.title()} {n}',
             email=f'{role}{n}@example.com',
             role=role, active=True, password_set=True)
    u.set_password('pw-correct1')
    for k, v in kw.items():
        setattr(u, k, v)
    db.session.add(u)
    db.session.commit()
    return u


def _project(name=None):
    from app import db
    from app.models.project import Project
    p = Project(name=name or f'Contract {_uniq()}', active=True)
    db.session.add(p)
    db.session.commit()
    return p


def _facility(project=None, name=None):
    from app import db
    from app.models.facility import Facility
    f = Facility(name=name or f'Facility {_uniq()}', active=True,
                 project_id=project.id if project else None)
    db.session.add(f)
    db.session.commit()
    return f


def _issue(facility=None, description='Test issue', **kw):
    from app import db
    from app.models.issue import Issue
    from app.utils.time_utils import now_eastern
    i = Issue(facility_id=facility.id if facility else None,
              description=description,
              severity=kw.pop('severity', 'medium'),
              status=kw.pop('status', 'open'),
              reported_at=kw.pop('reported_at', now_eastern()))
    for k, v in kw.items():
        setattr(i, k, v)
    db.session.add(i)
    db.session.commit()
    return i


def _link(a, b, link_type='related'):
    from app import db
    from app.models.issue import IssueLink
    link = IssueLink(issue_id=a.id, linked_issue_id=b.id, link_type=link_type)
    db.session.add(link)
    db.session.commit()
    return link


def _assign_inspector(user, project):
    from app import db
    from app.models.inspector_assignment import InspectorAssignment
    db.session.add(InspectorAssignment(user_id=user.id, project_id=project.id))
    db.session.commit()


def _assign_customer(user, project):
    from app import db
    from app.models.project import CustomerAssignment
    db.session.add(CustomerAssignment(user_id=user.id, project_id=project.id))
    db.session.commit()


def _login(client, user):
    return client.post('/auth/login',
                       data={'username': user.username, 'password': 'pw-correct1'},
                       follow_redirects=False)


def _link_count():
    from app.models.issue import IssueLink
    return IssueLink.query.count()


# ── Model semantics ──────────────────────────────────────────────────────────

def test_a_link_reads_differently_from_each_end(client):
    """The stored direction carries meaning for 'duplicate'."""
    a, b = _issue(description='dupe'), _issue(description='original')
    link = _link(a, b, 'duplicate')

    assert link.label_for(a.id) == 'Duplicate of'
    assert link.label_for(b.id) == 'Duplicated by'
    assert link.other_issue(a.id).id == b.id
    assert link.other_issue(b.id).id == a.id


def test_related_reads_the_same_from_both_ends(client):
    a, b = _issue(), _issue()
    link = _link(a, b, 'related')
    assert link.label_for(a.id) == 'Related to'
    assert link.label_for(b.id) == 'Related to'


def test_one_row_appears_on_both_issues(client):
    """all_links() merges the two storage directions into one list."""
    a, b = _issue(), _issue()
    _link(a, b)

    assert _link_count() == 1
    assert len(a.all_links()) == 1
    assert len(b.all_links()) == 1


def test_exists_between_is_direction_agnostic(client):
    """The UniqueConstraint only covers the stored direction; this covers both."""
    from app.models.issue import IssueLink

    a, b = _issue(), _issue()
    _link(a, b)

    assert IssueLink.exists_between(a.id, b.id)
    assert IssueLink.exists_between(b.id, a.id)
    assert not IssueLink.exists_between(a.id, _issue().id)


def test_deleting_an_issue_removes_its_links_from_both_sides(client):
    """A surviving link would render a dead row on the other issue's page."""
    from app import db

    a, b, c = _issue(), _issue(), _issue()
    _link(a, b)          # a is the source
    _link(c, a)          # a is the target
    assert _link_count() == 2

    db.session.delete(a)
    db.session.commit()

    assert _link_count() == 0


# ── Creating links through the route ─────────────────────────────────────────

def test_admin_can_link_two_issues(client):
    admin = _user('admin')
    a, b = _issue(), _issue()
    _login(client, admin)

    client.post(f'/issues/{a.id}/links',
                data={'link_type': 'duplicate', 'linked_issue_id': str(b.id)},
                follow_redirects=True)

    from app.models.issue import IssueLink
    link = IssueLink.query.one()
    assert (link.issue_id, link.linked_issue_id) == (a.id, b.id)
    assert link.link_type == 'duplicate'
    assert link.created_by == admin.id


def test_a_leading_hash_is_accepted(client):
    """People type the issue number the way it is displayed."""
    _login(client, _user('admin'))
    a, b = _issue(), _issue()

    client.post(f'/issues/{a.id}/links',
                data={'link_type': 'related', 'linked_issue_id': f'#{b.id}'},
                follow_redirects=True)

    assert _link_count() == 1


def test_linking_does_not_touch_either_issue(client):
    """A link is PURELY NAVIGATIONAL — no status, SLA, assignee or follower."""
    from app import db
    from app.models.issue import Issue, IssueFollower

    _login(client, _user('admin'))
    a = _issue(status='open', severity='high')
    b = _issue(status='in_progress', severity='low')
    before = [(i.id, i.status, i.severity, i.assigned_to, i.resolved_at,
               i.sla_notified) for i in (a, b)]

    client.post(f'/issues/{a.id}/links',
                data={'link_type': 'duplicate', 'linked_issue_id': str(b.id)},
                follow_redirects=True)

    # db.session.get, not Model.query.get — SQLAlchemy 2.x (CLAUDE.md rule 11).
    after = [(i.id, i.status, i.severity, i.assigned_to, i.resolved_at,
              i.sla_notified)
             for i in (db.session.get(Issue, a.id), db.session.get(Issue, b.id))]
    assert before == after
    assert IssueFollower.query.count() == 0


@pytest.mark.parametrize('payload', [
    {'link_type': 'related', 'linked_issue_id': 'not-a-number'},
    {'link_type': 'related', 'linked_issue_id': ''},
    {'link_type': 'nonsense', 'linked_issue_id': '1'},
    {'linked_issue_id': '1'},
])
def test_malformed_link_requests_are_rejected_without_error(client, payload):
    _login(client, _user('admin'))
    a = _issue()

    res = client.post(f'/issues/{a.id}/links', data=payload,
                      follow_redirects=True)

    assert res.status_code == 200
    assert _link_count() == 0


def test_an_issue_cannot_be_linked_to_itself(client):
    _login(client, _user('admin'))
    a = _issue()

    client.post(f'/issues/{a.id}/links',
                data={'link_type': 'related', 'linked_issue_id': str(a.id)},
                follow_redirects=True)

    assert _link_count() == 0


def test_the_same_pair_cannot_be_linked_twice_in_either_direction(client):
    """The reverse direction is the case the UniqueConstraint cannot catch."""
    _login(client, _user('admin'))
    a, b = _issue(), _issue()

    client.post(f'/issues/{a.id}/links',
                data={'link_type': 'related', 'linked_issue_id': str(b.id)},
                follow_redirects=True)
    client.post(f'/issues/{b.id}/links',
                data={'link_type': 'duplicate', 'linked_issue_id': str(a.id)},
                follow_redirects=True)

    assert _link_count() == 1


def test_unlinking_works_from_either_end(client):
    _login(client, _user('admin'))
    a, b = _issue(), _issue()
    link = _link(a, b)

    # From the far end — the row is stored on a, deleted from b's page.
    client.post(f'/issues/{b.id}/links/{link.id}/delete', follow_redirects=True)

    assert _link_count() == 0


def test_cannot_delete_a_link_between_two_other_issues(client):
    """The link must actually touch the issue named in the URL."""
    _login(client, _user('admin'))
    a, b, unrelated = _issue(), _issue(), _issue()
    link = _link(a, b)

    client.post(f'/issues/{unrelated.id}/links/{link.id}/delete',
                follow_redirects=True)

    assert _link_count() == 1


# ── Permission ───────────────────────────────────────────────────────────────

@pytest.mark.parametrize('role', ['admin', 'director', 'auditor'])
def test_issue_managers_may_link(client, role):
    _login(client, _user(role))
    a, b = _issue(), _issue()

    client.post(f'/issues/{a.id}/links',
                data={'link_type': 'related', 'linked_issue_id': str(b.id)},
                follow_redirects=True)

    assert _link_count() == 1


def test_the_assignee_may_link_their_own_issue(client):
    """_can_manage_links mirrors the page's can_edit, which includes the assignee."""
    project = _project()
    facility = _facility(project)
    inspector = _user('inspector')
    _assign_inspector(inspector, project)

    a = _issue(facility=facility, assigned_to=inspector.id)
    b = _issue(facility=facility)

    _login(client, inspector)
    client.post(f'/issues/{a.id}/links',
                data={'link_type': 'related', 'linked_issue_id': str(b.id)},
                follow_redirects=True)

    assert _link_count() == 1


def test_a_project_manager_may_not_link(client):
    """Widening this is a one-line change — but change view.html's can_edit too."""
    _login(client, _user('project_manager'))
    a, b = _issue(), _issue()

    res = client.post(f'/issues/{a.id}/links',
                      data={'link_type': 'related', 'linked_issue_id': str(b.id)})

    assert res.status_code == 403
    assert _link_count() == 0


def test_a_customer_may_not_link_even_on_their_own_facility(client):
    project = _project()
    facility = _facility(project)
    customer = _user('customer')
    _assign_customer(customer, project)

    a = _issue(facility=facility)
    b = _issue(facility=facility)

    _login(client, customer)
    res = client.post(f'/issues/{a.id}/links',
                      data={'link_type': 'related', 'linked_issue_id': str(b.id)})

    assert res.status_code == 403
    assert _link_count() == 0


# ── Scope: the part that actually matters ────────────────────────────────────

def test_cannot_link_to_an_issue_outside_your_scope(client):
    """An inspector must not be able to attach another contract's issue."""
    mine, theirs = _project('Mine'), _project('Theirs')
    my_facility, their_facility = _facility(mine), _facility(theirs)

    inspector = _user('inspector')
    _assign_inspector(inspector, mine)

    a = _issue(facility=my_facility, assigned_to=inspector.id)
    out_of_scope = _issue(facility=their_facility, description='another contract')

    _login(client, inspector)
    res = client.post(f'/issues/{a.id}/links',
                      data={'link_type': 'related',
                            'linked_issue_id': str(out_of_scope.id)},
                      follow_redirects=True)

    assert _link_count() == 0
    # And the refusal must not confirm the issue exists.
    assert 'another contract' not in res.get_data(as_text=True)


def test_the_panel_hides_a_link_whose_far_end_is_out_of_scope(client):
    """An admin can link across contracts. A customer at one end must still not
    read the issue at the other."""
    mine, theirs = _project('Mine'), _project('Theirs')
    my_facility, their_facility = _facility(mine), _facility(theirs)

    customer = _user('customer')
    _assign_customer(customer, mine)

    visible = _issue(facility=my_facility, description='my own issue')
    hidden = _issue(facility=their_facility,
                    description='SECRET other customer issue')
    _link(visible, hidden)

    _login(client, customer)
    body = client.get(f'/issues/{visible.id}').get_data(as_text=True)

    assert 'my own issue' in body
    assert 'SECRET other customer issue' not in body
    assert f'/issues/{hidden.id}' not in body


def test_link_search_is_scoped(client):
    mine, theirs = _project('Mine'), _project('Theirs')
    my_facility, their_facility = _facility(mine), _facility(theirs)

    inspector = _user('inspector')
    _assign_inspector(inspector, mine)

    a = _issue(facility=my_facility, description='mine mine mine')
    _issue(facility=my_facility, description='mine also')
    _issue(facility=their_facility, description='mine but theirs')

    _login(client, inspector)
    results = client.get(f'/issues/{a.id}/link-search?q=mine').get_json()['results']

    descriptions = {r['description'] for r in results}
    assert 'mine also' in descriptions
    assert 'mine but theirs' not in descriptions


def test_link_search_excludes_self_and_already_linked(client):
    _login(client, _user('admin'))
    a = _issue(description='alpha one')
    b = _issue(description='alpha two')
    _link(a, b)

    ids = {r['id'] for r in
           client.get(f'/issues/{a.id}/link-search?q=alpha').get_json()['results']}

    assert a.id not in ids
    assert b.id not in ids


def test_link_search_finds_by_issue_number(client):
    _login(client, _user('admin'))
    a = _issue(description='alpha')
    b = _issue(description='beta')

    ids = {r['id'] for r in
           client.get(f'/issues/{a.id}/link-search?q={b.id}').get_json()['results']}

    assert b.id in ids


def test_link_search_returns_nothing_for_an_unscoped_inspector(client):
    """Empty scope means no access to anything — rule 57's fail-closed default."""
    inspector = _user('inspector')          # no InspectorAssignment rows
    a = _issue(facility=_facility(_project()), assigned_to=inspector.id)

    _login(client, inspector)
    res = client.get(f'/issues/{a.id}/link-search?q=a')

    # Either the issue itself is unreachable (403) or the search finds nothing.
    assert res.status_code == 403 or res.get_json()['results'] == []


# ── Rendering ────────────────────────────────────────────────────────────────

def test_the_panel_renders_both_directions(client):
    _login(client, _user('admin'))
    original = _issue(description='the original')
    dupe = _issue(description='the duplicate')
    _link(dupe, original, 'duplicate')

    on_dupe = client.get(f'/issues/{dupe.id}').get_data(as_text=True)
    on_original = client.get(f'/issues/{original.id}').get_data(as_text=True)

    assert 'Duplicate of' in on_dupe
    assert f'/issues/{original.id}' in on_dupe
    assert 'Duplicated by' in on_original
    assert f'/issues/{dupe.id}' in on_original


def test_the_panel_renders_with_no_links(client):
    _login(client, _user('admin'))
    a = _issue()

    body = client.get(f'/issues/{a.id}').get_data(as_text=True)

    assert 'Linked Issues' in body
    assert 'No linked issues.' in body