gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: complete control panel fixture UX coverage

Browse Source
This commit is contained in:
gti
2026-09-22 20:49:04 -03:00
parent 0eeb7d159c
commit 4ba0fa8a53
4 changed files with 214 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
central/scripts/generate_control_panel_fixture.py
+16 -6
View File
@@ -23,16 +23,26 @@ def generate_fixture(session, agent_count=100, seed=7):
last_seen = None if state == "UNKNOWN" else now - timedelta(seconds=10 if state in {"ONLINE", "DEGRADED"} else 10_000)
installation = Installation(agent_id=agent.id, tenant_id=tenant.id, installation_id=str(uuid.uuid5(uuid.NAMESPACE_DNS, f"fixture-{index}")), display_name=f"{tenant.name} - Servidor Principal", customer_code=f"F{index:03d}", hostname=f"srv-{index:03d}", company_name=f"Empresa {index:03d}", company_uf="BA", company_crt="1", last_seen_at=last_seen, last_status="IDLE")
session.add(installation); session.flush()
snapshot = AgentSnapshot(installation_id=installation.id, total_products=5000 + index, fiscally_complete=4900 + index, safe_blocked=index % 9, failed_products=1 if state == "FAILED" else 0, coverage_percent=98.0, blockers={"CFOP": index % 7, "ST": index % 5}, agent_health=state, version_drift="UNKNOWN" if agent_version is None else ("OUTDATED" if agent_version != "1.0.0rc1" else "CURRENT"), last_job_status="FAILED" if state in {"DEGRADED", "FAILED"} else "COMPLETED")
snapshot = AgentSnapshot(installation_id=installation.id, total_products=5000 + index, fiscally_complete=4900 + index, safe_blocked=1 if index % 2 == 0 else 0, failed_products=1 if state == "FAILED" else 0, coverage_percent=98.0, blockers={"CFOP": index % 7, "ST": index % 5}, agent_health=state, version_drift="UNKNOWN" if agent_version is None else ("OUTDATED" if agent_version != "1.0.0rc1" else "CURRENT"), last_job_status="FAILED" if state in {"DEGRADED", "FAILED"} else "COMPLETED")
session.add(snapshot); created["agents"] += 1
for event_index in range(5):
session.add(AgentEvent(tenant_id=tenant.id, installation_id=installation.id, agent_id=agent.id, severity="ERROR" if state == "FAILED" and event_index == 0 else "INFO", event_type="JOB_FAILED" if state in {"DEGRADED", "FAILED"} and event_index == 0 else "HEARTBEAT", message="Fixture operational event", structured_context={"fixture": True}, occurred_at=now - timedelta(minutes=event_index)))
created["events"] += 1
session.add(Job(agent_id=agent.id, kind="REPROCESS_ALL", status="FAILED" if state in {"DEGRADED", "FAILED"} else "COMPLETED", payload={}, result_summary={"fixture": True}, error="fixture technical failure" if state in {"DEGRADED", "FAILED"} else None))
failed = state in {"DEGRADED", "FAILED"}
job = Job(agent_id=agent.id, kind="REPROCESS_ALL", status="FAILED" if failed else "COMPLETED", payload={}, result_summary={"fixture": True}, error="fixture technical failure" if failed else None, started_at=now - timedelta(minutes=6) if failed else None, finished_at=now - timedelta(minutes=5) if failed else None)
session.add(job); session.flush()
created["jobs"] += 1
fixture_blocker = None
if index % 2 == 0:
session.add(BlockedProduct(tenant_id=tenant.id, installation_id=installation.id, company=tenant.name, product_code=f"P{index:06d}", description="Fixture product", blocker_type="CFOP", field="CFOP", reason_code="INSUFFICIENT_EVIDENCE", evidence_summary={"fixture": True}, input_fingerprint=f"{index:064d}", blocker_fingerprint=f"{index+1000:064d}"))
fixture_blocker = BlockedProduct(tenant_id=tenant.id, installation_id=installation.id, company=tenant.name, product_code=f"P{index:06d}", description="Fixture product", blocker_type="CFOP", field="CFOP", reason_code="INSUFFICIENT_EVIDENCE", evidence_summary={"fixture": True}, input_fingerprint=f"{index:064d}", blocker_fingerprint=f"{index+1000:064d}")
session.add(fixture_blocker); session.flush()
created["blockers"] += 1
if index == 0:
session.add(BlockedProduct(tenant_id=tenant.id, installation_id=installation.id, company=tenant.name, product_code=fixture_blocker.product_code, description="Fixture product", blocker_type="TAX_RULE_MISSING", field="ST", reason_code="MISSING_REFERENCE", evidence_summary={"fixture": True}, input_fingerprint=f"{index+2000:064d}", blocker_fingerprint=f"{index+3000:064d}"))
created["blockers"] += 1
for event_index in range(5):
linked = event_index == 0
session.add(AgentEvent(tenant_id=tenant.id, installation_id=installation.id, agent_id=agent.id, severity="ERROR" if failed and linked else "INFO", event_type="JOB_FAILED" if failed and linked else "HEARTBEAT", message="Fixture job failed" if failed and linked else "Fixture operational event", structured_context={"fixture": True}, job_id=job.id if linked else None, product_code=fixture_blocker.product_code if linked and fixture_blocker else None, blocker_id=fixture_blocker.id if linked and fixture_blocker else None, occurred_at=now - timedelta(minutes=event_index)))
created["events"] += 1
session.commit()
return created
central/tests/test_control_panel_fixture.py
+33 -5
View File
@@ -1,15 +1,43 @@
from central.tests.test_control_plane import setup_db
import uuid
from central.scripts.generate_control_panel_fixture import generate_fixture
from central.app.models import AgentEvent, AgentSnapshot, Installation, Job
from central.app.models import AgentEvent, AgentSnapshot, BlockedProduct, Installation, Job
from central.tests.test_control_plane import setup_db
def test_fixture_creates_one_hundred_varied_installations_and_events():
session_factory = setup_db()
with session_factory() as session:
result = generate_fixture(session, agent_count=100, seed=11)
assert result == {"agents": 100, "events": 500, "blockers": 50, "jobs": 100}
assert result == {"agents": 100, "events": 500, "blockers": 51, "jobs": 100}
assert session.query(Installation).count() == 100
assert session.query(AgentEvent).count() == 500
assert session.query(Job).count() == 100
assert {row.agent_health for row in session.query(AgentSnapshot).all()} == {"ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN"}
assert {row.version_drift for row in session.query(AgentSnapshot).all()} == {"CURRENT", "OUTDATED", "UNKNOWN"}
snapshots = session.query(AgentSnapshot).all()
assert {row.agent_health for row in snapshots} == {"ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN"}
assert {row.version_drift for row in snapshots} == {"CURRENT", "OUTDATED", "UNKNOWN"}
assert {row.installation_id for row in session.query(Installation).all()} == {str(uuid.uuid5(uuid.NAMESPACE_DNS, f"fixture-{index}")) for index in range(100)}
blockers = session.query(BlockedProduct).all()
assert len({row.blocker_fingerprint for row in blockers}) == len(blockers)
for snapshot in snapshots:
product_codes = {row.product_code for row in blockers if row.installation_id == snapshot.installation_id and row.resolved_at is None}
assert snapshot.safe_blocked == len(product_codes)
repeated_product = next(row for row in blockers if row.product_code == "P000000")
same_product_blockers = [row for row in blockers if row.tenant_id == repeated_product.tenant_id and row.installation_id == repeated_product.installation_id and row.product_code == repeated_product.product_code]
assert len(same_product_blockers) >= 2
assert all(row.resolved_at is None for row in same_product_blockers)
assert all(row.tenant_id == session.query(Installation).filter_by(id=row.installation_id).one().tenant_id for row in blockers)
failed_jobs = session.query(Job).filter_by(status="FAILED").all()
assert failed_jobs and all(job.started_at is not None and job.finished_at is not None and job.finished_at >= job.started_at for job in failed_jobs)
linked_events = session.query(AgentEvent).filter(AgentEvent.job_id.is_not(None), AgentEvent.blocker_id.is_not(None)).all()
assert linked_events
for event in linked_events:
job = session.get(Job, event.job_id)
blocker = session.get(BlockedProduct, event.blocker_id)
installation = session.query(Installation).filter_by(id=blocker.installation_id).one()
assert event.installation_id == installation.id == session.query(Installation).filter_by(agent_id=job.agent_id).one().id
assert event.tenant_id == blocker.tenant_id == installation.tenant_id
assert event.product_code == blocker.product_code
assert all("password" not in event.message.lower() and "token" not in event.message.lower() for event in session.query(AgentEvent).all())