mirror of
https://github.com/kharonsec/br-acc
synced 2026-05-14 10:56:29 +02:00
393e7dc3f0
Auth: JWT auth with python-jose + passlib, invite-code registration, user model + 3 Cypher queries, auth router, owner-scoped investigations. Rate limiting: slowapi on auth endpoints. Integration tests: testcontainers-based tests for entity, graph, search. Deployment: docker-compose.prod.yml, Caddyfile, backup + deploy scripts, GitHub Actions deploy workflow, deploy docs. ETL rewrite: CNPJ pipeline handles real Receita Federal CSV layout (37 cols), chunked file reading, proper field mapping. Download + explore scripts. Test fixtures with real CSV samples. Frontend polish: Spinner component, responsive CSS improvements across all pages, better navigation, visual refinements. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from neo4j import AsyncDriver, AsyncGraphDatabase, AsyncSession
|
|
from testcontainers.neo4j import Neo4jContainer
|
|
|
|
from icarus.main import app
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def neo4j_container() -> Neo4jContainer: # type: ignore[misc]
|
|
"""Start a Neo4j container for integration tests."""
|
|
container = Neo4jContainer("neo4j:5-community")
|
|
container.start()
|
|
yield container # type: ignore[misc]
|
|
container.stop()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def neo4j_uri(neo4j_container: Neo4jContainer) -> str:
|
|
return neo4j_container.get_connection_url()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def neo4j_auth(neo4j_container: Neo4jContainer) -> tuple[str, str]:
|
|
return ("neo4j", neo4j_container.NEO4J_ADMIN_PASSWORD)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def neo4j_driver(
|
|
neo4j_uri: str, neo4j_auth: tuple[str, str]
|
|
) -> AsyncIterator[AsyncDriver]:
|
|
driver = AsyncGraphDatabase.driver(neo4j_uri, auth=neo4j_auth)
|
|
# Apply schema
|
|
schema_path = Path(__file__).parent.parent.parent.parent / "infra" / "neo4j" / "init.cypher"
|
|
if schema_path.exists():
|
|
async with driver.session() as session:
|
|
for statement in schema_path.read_text().split(";"):
|
|
stmt = statement.strip()
|
|
if stmt and not stmt.startswith("//"):
|
|
await session.run(stmt)
|
|
# Seed dev data
|
|
seed_path = (
|
|
Path(__file__).parent.parent.parent.parent / "infra" / "scripts" / "seed-dev.cypher"
|
|
)
|
|
if seed_path.exists():
|
|
async with driver.session() as session:
|
|
for statement in seed_path.read_text().split(";"):
|
|
stmt = statement.strip()
|
|
if stmt and not stmt.startswith("//"):
|
|
await session.run(stmt)
|
|
yield driver
|
|
await driver.close()
|
|
|
|
|
|
@pytest.fixture
|
|
async def integration_session(neo4j_driver: AsyncDriver) -> AsyncIterator[AsyncSession]:
|
|
async with neo4j_driver.session() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
async def integration_client(neo4j_driver: AsyncDriver) -> AsyncIterator[AsyncClient]:
|
|
app.state.neo4j_driver = neo4j_driver
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|