"""The bounty report's link into the internal system registry. What the link buys: the finding's budget pot and a suggested owner stop being guesses. What it must never do: invent an owner, or let a broken link read as "nobody owns this". """ from __future__ import annotations import uuid import pytest from app.identity.person_models import IdentityPerson BOUNTY = "/api/v1/bounty" SYSTEMS = "/api/v1/systems" async def _person(db_session, *, name="Sys Owner"): person = IdentityPerson( id=uuid.uuid4(), work_email=f"o-{uuid.uuid4().hex[:8]}@t.com", display_name=name, employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:10]}", ) db_session.add(person) await db_session.flush() await db_session.commit() return person async def _system(client, headers, **overrides): payload = { "name": "PingCAP 官网", "slug": f"www-{uuid.uuid4().hex[:6]}", "kind": "marketing", "exposure": "public", "hostnames": ["www.pingcap.com"], "budget_track": "it", "in_bounty_scope": True, } payload.update(overrides) resp = await client.post(SYSTEMS, json=payload, headers=headers) assert resp.status_code == 200, resp.text return resp.json()["data"] async def _report(client, headers, **overrides): payload = { "title": "目录列举 on www", "researcher_email": "whitehat@example.com", "affected_system": "www.pingcap.com", } payload.update(overrides) resp = await client.post(f"{BOUNTY}/reports", json=payload, headers=headers) assert resp.status_code == 200, resp.text return resp.json()["data"] @pytest.mark.asyncio async def test_link_carries_owner_and_pot_into_the_report( client, admin_headers, db_session, ): owner = await _person(db_session) system = await _system( client, admin_headers, owner_person_id=str(owner.id), ) report = await _report(client, admin_headers, system_id=system["id"]) assert report["system_id"] == system["id"] assert report["system"]["name"] == "PingCAP 官网" assert report["system"]["owner"]["id"] == str(owner.id) # The pot follows the registered system rather than the platform # default: an IT finding charged to 产研 corrupts the budget report, # which is the whole reason there are two pots. assert report["budget_track"] == "it" @pytest.mark.asyncio async def test_explicit_track_beats_the_registered_one( client, admin_headers, ): system = await _system(client, admin_headers, budget_track="it") report = await _report( client, admin_headers, system_id=system["id"], budget_track="product", ) assert report["budget_track"] == "product" @pytest.mark.asyncio async def test_unlinked_report_uses_the_platform_default( client, admin_headers, ): report = await _report(client, admin_headers) assert report["system_id"] is None assert report["system"] is None assert report["budget_track"] == "product" @pytest.mark.asyncio async def test_link_to_an_unknown_system_is_rejected(client, admin_headers): """A dangling link would render as an empty owner and read as "nobody owns this" — a different and much more alarming claim than "not linked".""" resp = await client.post( f"{BOUNTY}/reports", json={ "title": "x", "researcher_email": "a@b.com", "system_id": str(uuid.uuid4()), }, headers=admin_headers, ) assert resp.status_code == 422 resp = await client.patch( f"{BOUNTY}/reports/{uuid.uuid4()}", json={"system_id": str(uuid.uuid4())}, headers=admin_headers, ) assert resp.status_code in (404, 422) @pytest.mark.asyncio async def test_link_can_be_corrected_and_cleared(client, admin_headers): first = await _system(client, admin_headers) second = await _system(client, admin_headers, name="Docs", kind="marketing") report = await _report(client, admin_headers, system_id=first["id"]) resp = await client.patch( f"{BOUNTY}/reports/{report['id']}", json={"system_id": second["id"]}, headers=admin_headers, ) assert resp.json()["data"]["system"]["name"] == "Docs" resp = await client.patch( f"{BOUNTY}/reports/{report['id']}", json={"system_id": None}, headers=admin_headers, ) data = resp.json()["data"] assert data["system_id"] is None # The researcher's own words survive the unlink — the free text was # never derived from the link. assert data["affected_system"] == "www.pingcap.com" @pytest.mark.asyncio async def test_reports_can_be_filtered_by_system(client, admin_headers): system = await _system(client, admin_headers) linked = await _report(client, admin_headers, system_id=system["id"]) await _report(client, admin_headers) rows = ( await client.get( f"{BOUNTY}/reports?system_id={system['id']}", headers=admin_headers, ) ).json()["data"] assert [row["id"] for row in rows] == [linked["id"]] @pytest.mark.asyncio async def test_owner_page_sees_the_system_name_but_not_its_owners( client, admin_headers, db_session, employee_user, employee_headers, ): """The assignee needs to recognise WHICH system. Handing them the registry's owner emails and budget pot on their confirmation page widens what one report exposes for no operational gain.""" me, _token = employee_user registered_owner = await _person(db_session, name="Someone Else") system = await _system( client, admin_headers, owner_person_id=str(registered_owner.id), ) report = await _report(client, admin_headers, system_id=system["id"]) assign = await client.post( f"{BOUNTY}/reports/{report['id']}/assign", json={"person_id": str(me.id)}, headers=admin_headers, ) assert assign.status_code == 200, assign.text detail = ( await client.get( f"/api/v1/me/bounty/{report['id']}", headers=employee_headers, ) ).json()["data"] assert detail["system"]["name"] == "PingCAP 官网" assert detail["system"]["exposure"] == "public" assert "owner" not in detail["system"] assert "budget_track" not in detail["system"]