772 lines
43 KiB
Python
772 lines
43 KiB
Python
from pathlib import Path
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from central.app.blocked_products import upsert_blocker
|
|
from central.app.events import record_event
|
|
from central.app.main import create_app
|
|
from central.app.models import Agent, AgentEvent, AgentSnapshot, Installation, Job, Tenant
|
|
from central.tests.test_control_plane import seed, setup_db
|
|
|
|
|
|
def _panel_client():
|
|
session_factory = setup_db()
|
|
agent_id, _ = seed(session_factory)
|
|
now = datetime(2026, 9, 22, 19, 11, tzinfo=timezone.utc)
|
|
with session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
agent = session.get(Agent, agent_id)
|
|
agent.agent_version = "0.9.0"
|
|
snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
|
|
snapshot.total_products = 50
|
|
snapshot.fiscally_complete = 49
|
|
snapshot.safe_blocked = 1
|
|
snapshot.failed_products = 0
|
|
snapshot.coverage_percent = 98
|
|
job = Job(
|
|
agent_id=agent_id,
|
|
kind="REPROCESS_PRODUCTS",
|
|
status="RUNNING",
|
|
created_at=now - timedelta(minutes=3),
|
|
started_at=now - timedelta(minutes=2),
|
|
payload={"codigos": ["P-001"]},
|
|
)
|
|
session.add(job)
|
|
session.flush()
|
|
blocker = upsert_blocker(
|
|
session,
|
|
installation.tenant_id,
|
|
{
|
|
"company": "Acme",
|
|
"product_code": "P-001",
|
|
"description": "Produto de teste",
|
|
"blocker_type": "BLOCKED_MISSING_EVIDENCE",
|
|
"field": "CFOP",
|
|
"reason_code": "MISSING_CONTEXT",
|
|
"input_fingerprint": "page-test",
|
|
"evidence_summary": {"source": "snapshot"},
|
|
},
|
|
installation_id=installation.id,
|
|
)
|
|
blocker.first_seen_at = now - timedelta(days=1, seconds=11)
|
|
blocker.last_seen_at = now.replace(second=37)
|
|
record_event(
|
|
session,
|
|
tenant_id=installation.tenant_id,
|
|
installation_id=installation.id,
|
|
agent_id=agent_id,
|
|
severity="WARNING",
|
|
event_type="PRODUCT_BLOCKED",
|
|
message="Contexto fiscal ausente",
|
|
job_id=job.id,
|
|
product_code="P-001",
|
|
blocker_id=blocker.id,
|
|
occurred_at=now,
|
|
)
|
|
session.commit()
|
|
|
|
client = TestClient(create_app(database_probe=lambda: True, session_factory=session_factory, bootstrap_token=None))
|
|
client._session_factory = session_factory
|
|
login = client.post("/login", data={"username": "operator", "password": "a-secure-password"}, follow_redirects=False)
|
|
assert login.status_code == 303
|
|
return client, agent_id, blocker.id
|
|
|
|
|
|
def test_dashboard_renders_operational_queue_labels_and_human_dates():
|
|
client, _, _ = _panel_client()
|
|
|
|
response = client.get("/")
|
|
|
|
assert response.status_code == 200
|
|
for label in ("Servidores", "Online", "Atenção", "Offline", "Falhou", "Não informado"):
|
|
assert label in response.text
|
|
assert "Produtos com blockers abertos" in response.text
|
|
assert "Blockers abertos" in response.text
|
|
assert "Atenção necessária" in response.text
|
|
assert "Ver blockers" in response.text
|
|
assert "Jobs com falha" in response.text
|
|
assert "Agents desatualizados" in response.text
|
|
assert not re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", response.text)
|
|
|
|
|
|
def test_dashboard_coverage_displays_reconciled_product_buckets():
|
|
client, _, _ = _panel_client()
|
|
|
|
response = client.get("/")
|
|
|
|
assert response.status_code == 200
|
|
for label in ("Produtos avaliados", "Completos", "Bloqueados com segurança", "Falhas técnicas", "Não classificados"):
|
|
assert label in response.text
|
|
assert "Produtos com blockers abertos" in response.text
|
|
assert "Cobertura reconciliada" in response.text
|
|
|
|
|
|
def test_agents_scope_querystring_distinguishes_active_inventory_from_history():
|
|
client, _, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
tenant = session.scalar(select(Tenant).where(Tenant.name == "Acme"))
|
|
tenant.active = False
|
|
session.commit()
|
|
|
|
current = client.get("/agents?scope=active")
|
|
history = client.get("/agents?scope=all")
|
|
|
|
assert current.status_code == history.status_code == 200
|
|
assert "Acme ERP" not in current.text
|
|
assert "Acme ERP" in history.text
|
|
|
|
|
|
def test_blockers_and_jobs_scope_routes_keep_history_and_server_filters():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
agent = session.get(Agent, agent_id)
|
|
agent.enabled = False
|
|
session.commit()
|
|
|
|
assert "P-001" not in client.get("/blockers?scope=active&field=CFOP&reason_code=MISSING_CONTEXT").text
|
|
assert "P-001" in client.get("/blockers?scope=all&field=CFOP&reason_code=MISSING_CONTEXT").text
|
|
assert "Acme ERP" not in client.get("/jobs?scope=active&status=RUNNING").text
|
|
assert "Acme ERP" in client.get("/jobs?scope=all&status=RUNNING").text
|
|
|
|
|
|
def test_dashboard_links_reach_scoped_health_blocker_failed_job_and_outdated_filters():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
upsert_blocker(session, installation.tenant_id, {"company": "Acme", "product_code": "P-NCM", "description": "Mouse GTI", "blocker_type": "BLOCKED_MISSING_EVIDENCE", "field": "NCM", "reason_code": "INVALID_NCM", "input_fingerprint": "ncm-link"}, installation_id=installation.id)
|
|
session.add(Job(agent_id=agent_id, kind="SYNC_INCREMENTAL", status="FAILED", created_at=datetime.now(timezone.utc), error="falha longa"))
|
|
session.commit()
|
|
|
|
home = client.get("/").text
|
|
|
|
assert 'href="/agents?scope=active"' in home
|
|
assert 'href="/agents?scope=active&health=ONLINE"' in home
|
|
assert 'href="/blockers?scope=active&status=OPEN&field=NCM"' in home
|
|
assert 'href="/jobs?scope=active&status=FAILED"' in home
|
|
assert 'href="/agents?scope=active&version=outdated"' in home
|
|
|
|
|
|
def test_list_filters_are_visible_and_clearable_with_ptbr_statuses():
|
|
client, _, _ = _panel_client()
|
|
|
|
agents = client.get("/agents?scope=active&health=ONLINE").text
|
|
blockers = client.get("/blockers?scope=active&field=CFOP").text
|
|
jobs = client.get("/jobs?scope=active&status=RUNNING").text
|
|
|
|
assert 'name="health"' in agents and 'value="ONLINE" selected' in agents
|
|
assert 'href="/agents?scope=active"' in agents and "Limpar filtros" in agents
|
|
assert "Saúde técnica: Online" not in agents
|
|
assert 'name="field"' in blockers and 'value="CFOP" selected' in blockers
|
|
assert "Evidência fiscal ausente" in blockers and "Prioridade: HIGH" not in blockers
|
|
assert 'href="/blockers?scope=active"' in blockers and "Limpar filtros" in blockers
|
|
assert 'value="RUNNING" selected' in jobs and "Executando" in jobs
|
|
assert 'href="/jobs?scope=active"' in jobs and "Limpar filtros" in jobs
|
|
|
|
|
|
def test_login_has_centered_card_and_explicit_field_labels():
|
|
client, _, _ = _panel_client()
|
|
|
|
response = client.get("/login")
|
|
|
|
assert response.status_code == 200
|
|
assert 'class="login-page"' in response.text
|
|
# Same card component as the GTI Gerenciador login (gti-theme.css .cartao).
|
|
assert 'class="cartao login-card"' in response.text
|
|
assert '<label for="login-username">Usuário</label>' in response.text
|
|
assert '<label for="login-password">Senha</label>' in response.text
|
|
|
|
|
|
def test_dashboard_lists_bounded_failed_job_rows_with_identity_error_age_and_empty_state():
|
|
client, agent_id, _ = _panel_client()
|
|
factory = client._session_factory
|
|
with factory() as session:
|
|
for index in range(12):
|
|
session.add(Job(agent_id=agent_id, kind="REPROCESS_PRODUCTS", status="FAILED", error=f"erro curto {index}", created_at=datetime(2026, 9, 22, 18, 0, index, tzinfo=timezone.utc), finished_at=datetime(2026, 9, 22, 18, 5, index, tzinfo=timezone.utc)))
|
|
session.commit()
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert "Acme" in response.text and "Acme ERP" in response.text and "srv-01" in response.text
|
|
assert "erro curto" in response.text and "Falhas recentes" in response.text and "Falhou" in response.text
|
|
assert 'href="/jobs?scope=active&status=FAILED&job_id=' in response.text
|
|
assert response.text.count('class="error-text job-error-summary"') == 10 # bounded, one clamped summary per row
|
|
assert 'title="2026-09-22T18:05:11+00:00"' in response.text
|
|
assert ">22/09/2026 15:05<" in response.text
|
|
with factory() as session:
|
|
session.query(Job).filter(Job.status == "FAILED").delete()
|
|
session.commit()
|
|
assert "Nenhum job com falha." in client.get("/").text
|
|
|
|
|
|
def test_blocker_summary_uses_complete_filtered_relation_and_tenant_product_identity():
|
|
client, agent_id, _ = _panel_client()
|
|
factory = client._session_factory
|
|
with factory() as session:
|
|
original = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
other_tenant = Tenant(name="Outra empresa")
|
|
session.add(other_tenant); session.flush()
|
|
other_agent = Agent(tenant_id=other_tenant.id, name="agent-2", token_hash="y" * 64)
|
|
session.add(other_agent); session.flush()
|
|
other_installation = Installation(agent_id=other_agent.id, tenant_id=other_tenant.id, installation_id="22222222-2222-4222-8222-222222222222", display_name="Outra ERP", hostname="srv-02")
|
|
session.add(other_installation); session.flush()
|
|
for index in range(55):
|
|
upsert_blocker(session, original.tenant_id, {"company": "Acme", "product_code": f"P-{index:03}", "blocker_type": "BLOCKED_MISSING_EVIDENCE", "field": "CFOP", "reason_code": "MISSING", "input_fingerprint": f"many-{index}"}, installation_id=original.id)
|
|
upsert_blocker(session, other_tenant.id, {"company": "Outra", "product_code": "P-000", "blocker_type": "BLOCKED_MISSING_EVIDENCE", "field": "NCM", "reason_code": "MISSING", "input_fingerprint": "other-tenant"}, installation_id=other_installation.id)
|
|
resolved = upsert_blocker(session, original.tenant_id, {"company": "Acme", "product_code": "P-RES", "blocker_type": "BLOCKED_MISSING_EVIDENCE", "field": "ST", "reason_code": "MISSING", "input_fingerprint": "resolved"}, installation_id=original.id)
|
|
resolved.resolved_at = datetime.now(timezone.utc)
|
|
session.commit()
|
|
open_page = client.get("/blockers?status=OPEN&field=CFOP")
|
|
assert open_page.status_code == 200
|
|
assert '<span>Blockers abertos</span><strong>56</strong>' in open_page.text
|
|
assert '<span>Produtos afetados</span><strong>55</strong>' in open_page.text
|
|
assert '<span>Clientes afetados</span><strong>1</strong>' in open_page.text
|
|
resolved_page = client.get("/blockers?status=RESOLVED")
|
|
assert '<span>Blockers abertos</span><strong>0</strong>' in resolved_page.text
|
|
assert '<span>Produtos afetados</span><strong>1</strong>' in resolved_page.text
|
|
all_page = client.get("/blockers?status=ALL")
|
|
assert '<span>Blockers abertos</span><strong>57</strong>' in all_page.text
|
|
assert '<span>Produtos afetados</span><strong>57</strong>' in all_page.text
|
|
assert "Produtos afetados" in open_page.text and "Clientes afetados" in open_page.text
|
|
|
|
|
|
def test_agents_render_separate_health_and_fiscal_status_with_server_sort():
|
|
client, _, _ = _panel_client()
|
|
|
|
response = client.get("/agents?sort=worst_coverage&health=ONLINE&search=Acme")
|
|
|
|
assert response.status_code == 200
|
|
for label in ("Cliente / servidor", "Saúde", "Status fiscal", "Cobertura", "Produtos", "Versão", "Último contato"):
|
|
assert label in response.text
|
|
assert "Acme ERP" in response.text
|
|
assert "srv-01" in response.text
|
|
assert "98,0%" in response.text
|
|
assert 'name="health"' in response.text
|
|
assert 'value="Acme"' in response.text
|
|
assert 'value="worst_coverage"' in response.text
|
|
|
|
|
|
def test_agent_list_and_detail_show_nao_informado_when_snapshot_missing():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
session.query(AgentSnapshot).filter(AgentSnapshot.installation_id == installation.id).delete()
|
|
session.commit()
|
|
|
|
listing = client.get("/agents")
|
|
assert listing.status_code == 200
|
|
assert "None completos" not in listing.text
|
|
assert "None bloqueados" not in listing.text
|
|
assert ">None<" not in listing.text
|
|
assert "Não informado" in listing.text
|
|
|
|
detail = client.get(f"/agents/{agent_id}")
|
|
assert detail.status_code == 200
|
|
assert ">None<" not in detail.text
|
|
assert "None completos" not in detail.text
|
|
assert "Não informado" in detail.text
|
|
|
|
|
|
def test_agent_list_hides_technical_uuid_and_explains_remainder():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
technical_uuid = installation.installation_id
|
|
agent = session.get(Agent, agent_id)
|
|
agent.agent_version = None
|
|
snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
|
|
snapshot.version_drift = "UNKNOWN"
|
|
session.commit()
|
|
|
|
response = client.get("/agents")
|
|
|
|
assert response.status_code == 200
|
|
assert technical_uuid not in response.text
|
|
assert "bloqueados" in response.text
|
|
assert "Não informada" in response.text
|
|
assert "<td>Desconhecida<small>Desconhecida</small></td>" not in response.text
|
|
|
|
|
|
def test_agent_list_and_detail_show_localized_last_seen_with_iso_tooltip():
|
|
client, agent_id, _ = _panel_client()
|
|
expected_iso = '2026-09-22T19:11:37+00:00'
|
|
expected_display = '22/09/2026 16:11'
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
installation.last_seen_at = datetime(2026, 9, 22, 19, 11, 37, tzinfo=timezone.utc)
|
|
session.commit()
|
|
|
|
agents = client.get("/agents")
|
|
detail = client.get(f"/agents/{agent_id}")
|
|
|
|
assert agents.status_code == 200
|
|
assert f'title="{expected_iso}"' in agents.text
|
|
assert f'>{expected_display}<' in agents.text
|
|
assert detail.status_code == 200
|
|
assert f'title="{expected_iso}"' in detail.text
|
|
assert f'>{expected_display} ·' in detail.text
|
|
|
|
|
|
def test_jobs_render_client_identity_duration_and_preserve_filters():
|
|
client, _, _ = _panel_client()
|
|
|
|
response = client.get("/jobs?status=RUNNING&search=Acme")
|
|
|
|
assert response.status_code == 200
|
|
for label in ("Cliente / servidor", "Tipo / job", "Status", "Criado", "Finalizado / duração", "Erro / ação"):
|
|
assert label in response.text
|
|
assert "Acme" in response.text
|
|
assert "srv-01" in response.text
|
|
assert "em andamento" in response.text
|
|
assert 'value="Acme"' in response.text
|
|
assert '<option value="RUNNING" selected>' in response.text
|
|
assert "Agent 1" not in response.text
|
|
|
|
|
|
def test_jobs_show_localized_created_and_finished_with_iso_tooltips():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
session.add(Job(agent_id=agent_id, kind="REPROCESS_PRODUCTS", status="COMPLETED", created_at=datetime(2026, 9, 22, 19, 8, tzinfo=timezone.utc), started_at=datetime(2026, 9, 22, 19, 9, tzinfo=timezone.utc), finished_at=datetime(2026, 9, 22, 19, 10, tzinfo=timezone.utc)))
|
|
session.commit()
|
|
|
|
response = client.get("/jobs")
|
|
|
|
assert response.status_code == 200
|
|
assert 'title="2026-09-22T19:08:00+00:00"' in response.text
|
|
assert 'title="2026-09-22T19:10:00+00:00"' in response.text
|
|
assert ">22/09/2026 16:08<" in response.text
|
|
assert ">22/09/2026 16:10<" in response.text
|
|
|
|
|
|
def test_blocker_list_and_detail_show_localized_seen_times_with_iso_tooltips():
|
|
client, _, blocker_id = _panel_client()
|
|
|
|
listing = client.get("/blockers")
|
|
detail = client.get(f"/blockers/{blocker_id}")
|
|
|
|
assert listing.status_code == 200
|
|
assert 'title="2026-09-22T19:11:37+00:00"' in listing.text
|
|
assert ">22/09/2026 16:11<" in listing.text
|
|
assert detail.status_code == 200
|
|
assert 'title="2026-09-21T19:10:49+00:00"' in detail.text
|
|
assert 'title="2026-09-22T19:11:37+00:00"' in detail.text
|
|
assert ">22/09/2026 16:11<" in detail.text
|
|
|
|
|
|
def test_blockers_render_visible_filters_summary_priority_and_detail_link():
|
|
client, _, blocker_id = _panel_client()
|
|
|
|
response = client.get("/blockers?client=Acme&installation=srv-01&field=CFOP&blocker_type=BLOCKED_MISSING_EVIDENCE&status=OPEN&age=0h&search=P-001")
|
|
|
|
assert response.status_code == 200
|
|
for label in ("Cliente", "Instalação / servidor", "Campo", "Tipo de blocker", "Status", "Idade", "Produto ou busca", "Blockers abertos", "Produtos afetados", "Clientes afetados", "CFOP", "ST", "ICMS", "CSOSN", "NCM", "Outros", "Prioridade"):
|
|
assert label in response.text
|
|
assert f"/blockers/{blocker_id}" in response.text
|
|
assert "Alta" in response.text and "Evidência fiscal ausente" in response.text
|
|
assert 'value="Acme"' in response.text
|
|
assert 'value="BLOCKED_MISSING_EVIDENCE"' in response.text
|
|
assert "window.confirm" not in response.text
|
|
|
|
|
|
def test_agent_and_blocker_details_keep_safe_forms_and_human_event_rows():
|
|
client, agent_id, blocker_id = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
upsert_blocker(session, installation.tenant_id, {"company": "Acme", "product_code": "LEGACY", "blocker_type": "BLOCKED_MISSING_EVIDENCE", "field": "ST", "reason_code": "LEGACY", "input_fingerprint": "tenant-only"})
|
|
session.commit()
|
|
|
|
agent_response = client.get(f"/agents/{agent_id}?tab=events&severity=WARNING&event_type=PRODUCT_BLOCKED&page=2")
|
|
blocker_response = client.get(f"/blockers/{blocker_id}")
|
|
|
|
assert agent_response.status_code == 200
|
|
for label in ("Cliente", "Nome de exibição", "Hostname", "Saúde técnica", "Status fiscal", "Último heartbeat", "Versão", "Empresa", "UF", "CRT", "Eventos recentes", "Severidade", "Tipo de evento"):
|
|
assert label in agent_response.text
|
|
assert "Contexto fiscal ausente" in agent_response.text
|
|
assert f"/blockers?agent_id={agent_id}" in agent_response.text
|
|
assert ">CFOP</span><b>1</b>" in agent_response.text
|
|
assert ">ST</span><b>0</b>" in agent_response.text
|
|
agent_blockers = client.get(f"/blockers?agent_id={agent_id}")
|
|
assert "Produto de teste" in agent_blockers.text and "LEGACY" not in agent_blockers.text
|
|
assert not re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", re.sub(r' title="[^"]*"', "", agent_response.text))
|
|
assert 'value="WARNING" selected' in agent_response.text
|
|
assert 'value="PRODUCT_BLOCKED"' in agent_response.text
|
|
|
|
assert blocker_response.status_code == 200
|
|
for label in ("Detalhes do blocker", "Evidência", "Contexto atual", "Salvar contexto e reprocessar", "Chave de contexto", "Valor", "Escopo", "Nota"):
|
|
assert label in blocker_response.text
|
|
assert 'name="context"' not in blocker_response.text
|
|
assert "data-confirm" not in blocker_response.text
|
|
assert not re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", re.sub(r' title="[^"]*"', "", blocker_response.text))
|
|
|
|
|
|
def test_legacy_blocker_detail_explains_unavailable_resolution_without_form():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
blocker = upsert_blocker(
|
|
session,
|
|
installation.tenant_id,
|
|
{"company": "Acme", "product_code": "LEGACY", "blocker_type": "BLOCKED_MISSING_EVIDENCE", "field": "ST", "reason_code": "LEGACY", "input_fingerprint": "unbound-detail"},
|
|
)
|
|
session.commit()
|
|
blocker_id = blocker.id
|
|
|
|
response = client.get(f"/blockers/{blocker_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "blocker legado de empresa" in response.text.lower()
|
|
assert "sem vínculo comprovado com um servidor" in response.text.lower()
|
|
assert "não é possível oferecer o reprocessamento de contexto com segurança" in response.text.lower()
|
|
assert "Salvar contexto e reprocessar" not in response.text
|
|
assert f'action="/blockers/{blocker_id}/resolve-context"' not in response.text
|
|
|
|
|
|
def test_agent_detail_resolves_agent_after_first_hundred_installations():
|
|
client, _, _ = _panel_client()
|
|
factory = client._session_factory
|
|
with factory() as session:
|
|
tenant = session.scalar(select(Tenant).where(Tenant.name == "Acme"))
|
|
target_id = 0
|
|
for index in range(105):
|
|
agent = Agent(tenant_id=tenant.id, name=f"late-{index}", token_hash=f"{index:064d}")
|
|
session.add(agent); session.flush()
|
|
session.add(Installation(agent_id=agent.id, tenant_id=tenant.id, installation_id=f"{index + 10:08d}-1111-4111-8111-111111111111", display_name=f"Late ERP {index}", hostname=f"late-{index}"))
|
|
target_id = agent.id
|
|
session.commit()
|
|
response = client.get(f"/agents/{target_id}")
|
|
assert response.status_code == 200
|
|
assert "Late ERP 104" in response.text
|
|
|
|
|
|
def test_agent_detail_reconciles_coverage_and_formats_missing_technical_fields():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
agent = session.get(Agent, agent_id)
|
|
agent.agent_version = None
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
|
|
snapshot.version_drift = "UNKNOWN"
|
|
session.commit()
|
|
|
|
response = client.get(f"/agents/{agent_id}")
|
|
|
|
assert response.status_code == 200
|
|
for label in ("Produtos", "Completos", "Bloqueados com segurança", "Falhas técnicas", "Não classificados", "Cobertura fiscal"):
|
|
assert label in response.text
|
|
assert "Não informado" in response.text
|
|
assert "Desconhecida" in response.text
|
|
assert "Desconhecida · Desconhecida" not in response.text
|
|
assert "ODS -" not in response.text
|
|
context = response.text.split("<h2>Contexto da empresa</h2>", 1)[1].split("</article>", 1)[0]
|
|
assert "Reprocessar tudo" not in context # the only reprocess action lives in the Processamento panel
|
|
assert response.text.count("Reprocessar tudo") == 1
|
|
assert not re.search(r"<(?:a|button)\b[^>]*>\s*</(?:a|button)>", context)
|
|
|
|
|
|
def test_agent_detail_reports_negative_coverage_counts_as_data_inconsistency():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
|
|
snapshot.failed_products = -1
|
|
session.commit()
|
|
|
|
response = client.get(f"/agents/{agent_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "contagens negativas" in response.text
|
|
|
|
|
|
def test_agent_event_filters_lookahead_and_safe_event_reference_links_and_confirmation():
|
|
client, agent_id, _ = _panel_client()
|
|
factory = client._session_factory
|
|
with factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
job = session.scalar(select(Job).where(Job.agent_id == agent_id))
|
|
for index in range(27):
|
|
session.add(AgentEvent(tenant_id=installation.tenant_id, installation_id=installation.id, agent_id=agent_id, severity="INFO", event_type="HEARTBEAT", message=f"evento {index}", structured_context={}, job_id=job.id, product_code="P-001", occurred_at=datetime(2026, 9, 21 + index % 2, tzinfo=timezone.utc)))
|
|
session.add(AgentEvent(tenant_id=installation.tenant_id, installation_id=installation.id, agent_id=agent_id, severity="INFO", event_type="HEARTBEAT", message="evento anterior ao filtro", structured_context={}, occurred_at=datetime(2026, 9, 20, tzinfo=timezone.utc)))
|
|
session.commit()
|
|
first = client.get(f"/agents/{agent_id}?tab=events&since=2026-09-21&until=2026-09-22")
|
|
assert first.status_code == 200
|
|
assert 'name="since"' in first.text and 'name="until"' in first.text
|
|
assert "evento anterior ao filtro" not in first.text
|
|
assert "Próxima" in first.text and "since=2026-09-21" in first.text and "until=2026-09-22" in first.text
|
|
last = client.get(f"/agents/{agent_id}?tab=events&event_page=2&since=2026-09-21&until=2026-09-22")
|
|
assert "Anterior" in last.text and "Próxima" not in last.text
|
|
assert "/jobs?job_id=" in first.text and "/blockers?agent_id=" in first.text
|
|
csrf = client.cookies.get("gti_csrf")
|
|
blocked = client.post(f"/agents/{agent_id}/reprocess", data={"csrf_token": csrf})
|
|
assert blocked.status_code == 400 and "confirmar" in blocked.text.lower() and 'name="confirm"' in blocked.text
|
|
confirmation = client.get(f"/agents/{agent_id}/reprocess-confirmation")
|
|
assert confirmation.status_code == 200 and 'name="confirm"' in confirmation.text and 'required' in confirmation.text
|
|
assert f'href="/agents/{agent_id}"' in confirmation.text
|
|
assert client.post(f"/agents/{agent_id}/reprocess", data={"csrf_token": "wrong", "confirm": "yes"}).status_code == 403
|
|
assert client.post(f"/agents/{agent_id}/reprocess", data={"csrf_token": csrf, "confirm": "yes"}, follow_redirects=False).status_code == 303
|
|
|
|
|
|
def test_theme_defaults_to_dark_and_never_overrides_a_saved_choice():
|
|
client, _, _ = _panel_client()
|
|
base = client.get("/agents").text
|
|
# no saved preference -> dark; an explicit "claro" is the only thing that yields light
|
|
assert 'tema === "claro" ? "claro" : "escuro"' in base
|
|
assert "prefers-color-scheme" not in (Path(__file__).parents[1] / "static" / "tema.js").read_text(encoding="utf-8")
|
|
script = (Path(__file__).parents[1] / "static" / "tema.js").read_text(encoding="utf-8")
|
|
assert 'return "escuro"' in script and 'localStorage.setItem("tema", novo)' in script
|
|
for page in ("login.html", "logged_out.html", "sso_error.html"):
|
|
assert 'data-tema="escuro"' in (Path(__file__).parents[1] / "templates" / page).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_operator_pages_use_portuguese_titles_and_consistent_ctas():
|
|
client, agent_id, _ = _panel_client()
|
|
assert "<title>Agents / Servidores | GTI Fiscal</title>" in client.get("/agents").text
|
|
assert "<title>Blockers | GTI Fiscal</title>" in client.get("/blockers").text
|
|
assert "<title>Jobs | GTI Fiscal</title>" in client.get("/jobs").text
|
|
dashboard = client.get("/").text
|
|
for old in ("Abrir fila", "Ver instalações", "Ver todos", "Ver job #"):
|
|
assert old not in dashboard
|
|
|
|
|
|
def test_agent_detail_processing_hierarchy_and_status_badges():
|
|
client, agent_id, _ = _panel_client()
|
|
page = client.get(f"/agents/{agent_id}").text
|
|
assert page.count("Reprocessar tudo") == 1
|
|
assert 'class="primary-action"' in page and 'class="botao secondary"' in page and 'class="caution"' in page
|
|
assert "Mantém o heartbeat, mas não executa jobs" in page and "Recalcula todos os produtos" in page
|
|
assert "Saúde técnica: " not in page and "Status fiscal: " not in page # context lives in the small label, not in the badge
|
|
assert '<small class="status-label">Saúde técnica</small>' in page
|
|
|
|
|
|
def _complete_sync(client, agent_id, fiscal_context):
|
|
with client._session_factory() as session:
|
|
session.add(Job(agent_id=agent_id, kind="SYNC_INCREMENTAL", status="COMPLETED", payload={}, result_summary={"fiscal_context": fiscal_context}, created_at=datetime(2026, 9, 22, 19, 0, tzinfo=timezone.utc),
|
|
finished_at=datetime(2026, 9, 22, 19, 5, tzinfo=timezone.utc)))
|
|
session.commit()
|
|
|
|
|
|
def test_agent_detail_has_one_coverage_and_shows_rtc_status_only_when_reported():
|
|
client, agent_id, _ = _panel_client()
|
|
plain = client.get(f"/agents/{agent_id}").text
|
|
assert "Reforma Tributária" not in plain # older Agents do not report it
|
|
for text in (plain,):
|
|
assert "Escopo anterior" not in text and "Escopo ampliado" not in text and "aplicabilidade desconhecida" not in text
|
|
_complete_sync(client, agent_id, {"rtc_enabled": False, "rtc_supported": True, "rtc_catalog_valid": True})
|
|
off = client.get(f"/agents/{agent_id}").text
|
|
assert re.search(r"Reforma Tributária</dt><dd><span[^>]*>Desativada</span>", off) and "Escopo anterior" not in off and "Escopo ampliado" not in off
|
|
assert "cClassTrib" not in off # RTC off never surfaces RTC as a problem
|
|
_complete_sync(client, agent_id, {"rtc_enabled": True, "rtc_supported": True, "rtc_catalog_valid": True})
|
|
on = client.get(f"/agents/{agent_id}").text
|
|
assert re.search(r"Reforma Tributária</dt><dd><span[^>]*>Ativa</span>", on)
|
|
|
|
|
|
def test_pis_cofins_and_rtc_blockers_appear_in_the_same_queue_and_dashboard_breakdown():
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
for code, field, reason in (("P-010", "PIS", "PIS_CST_INVALID"), ("P-011", "COFINS", "COFINS_CST_MISSING"), ("P-012", "RTC", "CCLASSTRIB_MISSING")):
|
|
upsert_blocker(session, installation.tenant_id, {"company": "Acme", "product_code": code, "description": "x", "blocker_type": "BLOCKED_FISCAL_DATA", "field": field, "reason_code": reason, "input_fingerprint": code},
|
|
installation_id=installation.id)
|
|
dash = client.get("/").text
|
|
for key in ("PIS", "COFINS", "RTC", "CEST"):
|
|
assert f"<span>{key}</span>" in dash
|
|
queue = client.get("/blockers?scope=active&status=OPEN").text
|
|
assert "CST PIS inválido" in queue and "CST COFINS ausente" in queue and "cClassTrib ausente" in queue
|
|
assert "fila RTC" not in queue.lower()
|
|
filtered = client.get("/blockers?scope=active&status=OPEN&field=RTC").text
|
|
assert "cClassTrib ausente" in filtered and "CST PIS inválido" not in filtered
|
|
|
|
|
|
def test_rtc_status_shows_where_and_when_the_agent_observed_it():
|
|
client, agent_id, _ = _panel_client()
|
|
_complete_sync(client, agent_id, {"rtc_enabled": True, "rtc_supported": True, "rtc_catalog_valid": True, "rtc_source": "PARAMETROS.USAR_REFORMA_TRIBUTARIA", "rtc_observed_at": "2026-09-24T20:00:00+00:00"})
|
|
text = client.get(f"/agents/{agent_id}").text
|
|
assert 'title="Lido no banco do cliente (PARAMETROS.USAR_REFORMA_TRIBUTARIA) em ' in text and ">Ativa<" in text
|
|
_complete_sync(client, agent_id, {"rtc_enabled": False, "rtc_param_conflict": True, "rtc_source": "PARAMETROS.USAR_REFORMA_TRIBUTARIA"})
|
|
conflict = client.get(f"/agents/{agent_id}").text
|
|
assert "linhas de PARAMETROS divergentes" in conflict and ">Desativada<" in conflict
|
|
for legacy in ("Escopo anterior", "Escopo ampliado", "Legacy coverage", "Expanded coverage"):
|
|
assert legacy not in conflict
|
|
|
|
|
|
# ---------------------------------------------------------------- RTC auto-apply authorization (central, default OFF)
|
|
def _installation(client, agent_id):
|
|
from central.app.models import Installation as Inst
|
|
with client._session_factory() as session:
|
|
row = session.scalar(select(Inst).where(Inst.agent_id == agent_id))
|
|
return row.rtc_write_mode
|
|
|
|
|
|
def test_rtc_write_mode_is_disabled_by_default_and_needs_csrf_valid_mode_and_a_typed_confirmation_for_automatic():
|
|
client, agent_id, _ = _panel_client()
|
|
assert _installation(client, agent_id) == "DISABLED"
|
|
csrf = client.cookies.get("gti_csrf")
|
|
url = f"/agents/{agent_id}/rtc-mode"
|
|
assert client.post(url, data={"csrf_token": "wrong", "mode": "DRY_RUN"}).status_code == 403
|
|
assert client.post(url, data={"csrf_token": csrf, "mode": "ENABLED"}).status_code == 422
|
|
assert client.post(url, data={"csrf_token": csrf, "mode": "ENABLED", "confirm": "sim"}).status_code == 422
|
|
assert client.post(url, data={"csrf_token": csrf, "mode": "bogus"}).status_code == 422
|
|
assert _installation(client, agent_id) == "DISABLED"
|
|
for mode, extra in (("DRY_RUN", {}), ("APPROVAL_REQUIRED", {}), ("ENABLED", {"confirm": "preencher"}), ("DISABLED", {})):
|
|
assert client.post(url, data={"csrf_token": csrf, "mode": mode, **extra}, follow_redirects=False).status_code == 303
|
|
assert _installation(client, agent_id) == mode
|
|
from central.app.models import AuditEvent
|
|
with client._session_factory() as session:
|
|
actions = [a.action for a in session.scalars(select(AuditEvent).order_by(AuditEvent.id)).all() if a.action.startswith("RTC_")]
|
|
assert actions == ["RTC_WRITE_MODE"] * 4
|
|
|
|
|
|
def _seed_rtc(client, agent_id, n=120):
|
|
from central.app.models import Installation as Inst, RtcCandidate, RtcWriteAudit
|
|
with client._session_factory() as session:
|
|
inst = session.scalar(select(Inst).where(Inst.agent_id == agent_id))
|
|
inst.rtc_write_mode = "APPROVAL_REQUIRED"
|
|
for i in range(n):
|
|
session.add(RtcCandidate(tenant_id=inst.tenant_id, installation_id=inst.id, product_code=f"C{i:04d}", description=f"PRODUTO {i}", ncm="10063021", proposed="200003", cst_ibscbs="200", rule_id="RTC-R3", basis="RTC-R3:ANNEX", status="PENDING"))
|
|
audit = RtcWriteAudit(tenant_id=inst.tenant_id, installation_id=inst.id, agent_id=agent_id, product_code="A1", field="CCLASSTRIB", outcome="APPLIED", old_value=None, new_value="200003", rule_id="RTC-R3", action="APPLY")
|
|
session.add(audit)
|
|
session.commit()
|
|
return audit.id
|
|
|
|
|
|
def test_rtc_control_page_paginates_candidates_on_the_server_and_shows_modes_counts_and_audit():
|
|
client, agent_id, _ = _panel_client()
|
|
_seed_rtc(client, agent_id)
|
|
page1 = client.get(f"/agents/{agent_id}/rtc").text
|
|
assert "Somente cClassTrib com regra fiscal determinística e catálogo validado pode ser preenchido automaticamente." in page1
|
|
for label in ("Desativada", "Dry-run", "Aprovação manual", "Automática"):
|
|
assert label in page1
|
|
assert "Pendentes (120)" in page1 and "C0000" in page1 and "C0049" in page1 and "C0050" not in page1 and "Página 1 de 3" in page1 and "Aplicar aprovados agora" in page1
|
|
page3 = client.get(f"/agents/{agent_id}/rtc?page=3").text
|
|
assert "C0119" in page3 and "C0099" not in page3 and "Página 3 de 3" in page3
|
|
assert "Reverter" in page3
|
|
|
|
|
|
def test_rtc_candidate_actions_approve_reject_apply_and_rollback_are_csrf_protected_and_scoped():
|
|
from central.app.models import Job, RtcCandidate, RtcWriteAudit
|
|
client, agent_id, _ = _panel_client()
|
|
audit_id = _seed_rtc(client, agent_id, n=10)
|
|
csrf = client.cookies.get("gti_csrf")
|
|
url = f"/agents/{agent_id}/rtc-candidates"
|
|
assert client.post(url, data={"csrf_token": "wrong", "action": "approve"}).status_code == 403
|
|
with client._session_factory() as session:
|
|
ids = [c.id for c in session.scalars(select(RtcCandidate).order_by(RtcCandidate.id)).all()]
|
|
assert client.post(url, content=f"csrf_token={csrf}&action=approve&ids={ids[0]}&ids={ids[1]}", headers={"content-type": "application/x-www-form-urlencoded"}, follow_redirects=False).status_code == 303
|
|
assert client.post(f"{url}?all=1", data={"csrf_token": csrf, "action": "approve"}, follow_redirects=False).status_code == 303
|
|
assert client.post(url, content=f"csrf_token={csrf}&action=reject&ids={ids[2]}", headers={"content-type": "application/x-www-form-urlencoded"}, follow_redirects=False).status_code == 303
|
|
with client._session_factory() as session:
|
|
counts = {c.product_code: c.status for c in session.scalars(select(RtcCandidate)).all()}
|
|
assert list(counts.values()).count("REJECTED") == 1 and list(counts.values()).count("APPROVED") == 9
|
|
jobs_before = len(session.scalars(select(Job)).all())
|
|
assert client.post(url, data={"csrf_token": csrf, "action": "apply_approved"}, follow_redirects=False).status_code == 303
|
|
with client._session_factory() as session:
|
|
job = session.scalars(select(Job).order_by(Job.id.desc())).first()
|
|
assert len(session.scalars(select(Job)).all()) == jobs_before + 1 and job.kind == "REPROCESS_PRODUCTS" and job.payload["reason"] == "RTC_APPLY_APPROVED" and len(job.payload["codigos"]) == 9
|
|
assert client.post(url, data={"csrf_token": csrf, "action": "nope"}).status_code == 422
|
|
roll = f"/agents/{agent_id}/rtc-rollback"
|
|
assert client.post(roll, data={"csrf_token": csrf, "audit_id": audit_id}).status_code == 422 # typed confirmation required
|
|
assert client.post(roll, data={"csrf_token": csrf, "audit_id": audit_id, "confirm": "reverter"}, follow_redirects=False).status_code == 303
|
|
assert client.post(roll, data={"csrf_token": csrf, "audit_id": audit_id, "confirm": "REVERTER"}).status_code == 409 # already requested: no duplicate
|
|
with client._session_factory() as session:
|
|
assert session.get(RtcWriteAudit, audit_id).rollback_state == "REQUESTED"
|
|
|
|
|
|
def test_apply_approved_is_refused_outside_the_approval_mode():
|
|
client, agent_id, _ = _panel_client()
|
|
csrf = client.cookies.get("gti_csrf")
|
|
assert client.post(f"/agents/{agent_id}/rtc-candidates", data={"csrf_token": csrf, "action": "apply_approved"}).status_code == 409
|
|
|
|
|
|
def test_panel_shows_rtc_products_summary_and_the_pending_message_only_when_rtc_is_on():
|
|
client, agent_id, _ = _panel_client()
|
|
_complete_sync(client, agent_id, {"rtc_enabled": False, "rtc_supported": True, "rtc_resolution": None})
|
|
off = client.get(f"/agents/{agent_id}").text
|
|
assert "Produtos RTC" not in off and "cClassTrib pendente" not in off and "Preenchimento automático" not in off
|
|
resolution = {"total": 521, "valid_existing": 0, "ready_to_apply": 280, "ambiguous": 215, "insufficient": 26, "catalog_compatibility": {"compatibility": "PARTIAL_LOCAL_OUTDATED"}}
|
|
_complete_sync(client, agent_id, {"rtc_enabled": True, "rtc_supported": True, "rtc_catalog_valid": True, "rtc_resolution": resolution})
|
|
on = client.get(f"/agents/{agent_id}").text
|
|
assert "Reforma Tributária ativa — cClassTrib pendente em 521 produtos" in on and "prontos para preenchimento 280" in on and "ambíguos/bloqueados 241" in on
|
|
assert "Preenchimento automático" in on and "Desativada" in on and f"/agents/{agent_id}/rtc" in on
|
|
for legacy in ("Escopo anterior", "Escopo ampliado"):
|
|
assert legacy not in on
|
|
|
|
|
|
def test_rtc_control_page_shows_human_labels_and_a_catalog_column_never_raw_enums():
|
|
from central.app.panel_presentation import abort_reason_label, audit_outcome_label, candidate_status_label, catalog_state_label, blocker_reason_label, blocker_type_label, rtc_rule_label
|
|
assert candidate_status_label("APPROVED") == "Aprovado" and audit_outcome_label("ROLLED_BACK") == "Revertido"
|
|
assert rtc_rule_label("RTC-R4") == "Sugestão sem autorização de preenchimento (RTC-R4)"
|
|
assert blocker_reason_label("INSUFFICIENT_EVIDENCE") == "Evidência fiscal insuficiente"
|
|
assert blocker_type_label("CCLASSTRIB_CONFLICT") == "cClassTrib existente diverge da classificação proposta"
|
|
assert catalog_state_label("MISSING_LOCAL") == "Classe oficial ausente no catálogo local do SGV"
|
|
assert abort_reason_label("VALUE_ALREADY_SET_NEVER_OVERWRITTEN") == "cClassTrib existente diverge da classificação proposta"
|
|
client, agent_id, _ = _panel_client()
|
|
page = client.get(f"/agents/{agent_id}/rtc").text
|
|
for header in ("cClassTrib atual", "cClassTrib proposto", "Fundamento", "Catálogo", "Situação", "Descrição"):
|
|
assert header in page
|
|
for raw in ("PENDING</td>", "APPLIED</td>", "ABORTED"):
|
|
assert raw not in page
|
|
|
|
|
|
def test_rtc_no_r3_candidates_shows_simples_explanation_and_filtered_blocker_link():
|
|
from central.app.models import BlockedProduct, RtcCandidate, RtcWriteAudit
|
|
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
installation.rtc_write_mode = "ENABLED"
|
|
session.add(RtcCandidate(tenant_id=installation.tenant_id, installation_id=installation.id, product_code="R4-SUGGESTION", description="Sugestão", ncm="10063021", proposed="000001", cst_ibscbs="000", rule_id="RTC-R4", basis="suggestion only", status="PENDING"))
|
|
session.add_all([
|
|
BlockedProduct(
|
|
tenant_id=installation.tenant_id,
|
|
installation_id=installation.id,
|
|
company="Acme",
|
|
product_code=f"RTC-{index:04d}",
|
|
description=f"Produto RTC {index}",
|
|
blocker_type="BLOCKED_FISCAL_DATA",
|
|
field="RTC",
|
|
reason_code="CCLASSTRIB_MISSING_UNRESOLVED",
|
|
evidence_summary={"reason": "RTC-R1:SIMPLES_ORIENTATION_PENDING_OFFICIAL_NT"},
|
|
input_fingerprint=f"rtc-input-{index}",
|
|
blocker_fingerprint=f"rtc-blocker-{index}",
|
|
first_seen_at=datetime.now(timezone.utc),
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
)
|
|
for index in range(521)
|
|
])
|
|
session.add(Job(agent_id=agent_id, kind="REPROCESS_ALL", status="COMPLETED", payload={}, result_summary={
|
|
"fiscal_context": {
|
|
"rtc_enabled": True,
|
|
"rtc_supported": True,
|
|
"rtc_catalog_valid": True,
|
|
"crt": "1",
|
|
"legal_effective_context": {"cclasstrib_orientation": "PENDING_OFFICIAL_NT"},
|
|
"rtc_resolution": {
|
|
"total": 521, "valid_existing": 0, "ready_to_apply": 0, "suggestion_only": 0,
|
|
"ambiguous": 0, "insufficient": 521,
|
|
"by_reason": {"SIMPLES_ORIENTATION_PENDING_OFFICIAL_NT": 521},
|
|
},
|
|
},
|
|
"rtc_auto_apply": {"mode": "ENABLED", "ready_total": 0, "preview": [], "write_attempts": 0, "write_successes": 0},
|
|
}, created_at=datetime.now(timezone.utc), finished_at=datetime.now(timezone.utc)))
|
|
session.commit()
|
|
|
|
page = client.get(f"/agents/{agent_id}/rtc")
|
|
|
|
assert page.status_code == 200
|
|
assert "Não há produtos elegíveis para preenchimento automático." in page.text
|
|
assert "521 produtos precisam de atenção fiscal, mas não podem ser preenchidos automaticamente pelas regras atuais." in page.text
|
|
assert "Esta empresa está no Simples Nacional." in page.text
|
|
assert "SIMPLES_ORIENTATION_PENDING_OFFICIAL_NT" not in page.text
|
|
assert "R4-SUGGESTION" not in page.text
|
|
assert 'href="/blockers?scope=all&agent_id=' + str(agent_id) + '&field=RTC&status=OPEN"' in page.text
|
|
assert "Nenhum candidato neste filtro." in page.text
|
|
|
|
blockers = client.get(f"/blockers?scope=all&agent_id={agent_id}&field=RTC&status=OPEN")
|
|
assert blockers.status_code == 200
|
|
assert "521 blockers" in blockers.text
|
|
with client._session_factory() as session:
|
|
assert session.scalar(select(Installation.rtc_write_mode).where(Installation.agent_id == agent_id)) == "ENABLED"
|
|
assert session.query(RtcCandidate).filter_by(installation_id=installation.id).count() == 1
|
|
assert session.query(RtcCandidate).filter_by(installation_id=installation.id, rule_id="RTC-R4").count() == 1
|
|
assert session.query(RtcWriteAudit).filter_by(installation_id=installation.id).count() == 0
|