377 lines
15 KiB
Python
377 lines
15 KiB
Python
"""Tests for scanner parsers (CSV, MD, YAML, OpenAPI)."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.modules.scanner.infrastructure.parsers.csv_parser import CsvParser
|
|
|
|
DESIGN_DIR = Path("/workspace/arch-design-agent-skill-dashboard/design")
|
|
|
|
|
|
@pytest.fixture
|
|
def csv_parser():
|
|
return CsvParser()
|
|
|
|
|
|
# ── CSV Parser Tests ──
|
|
|
|
|
|
class TestCsvParserCapabilities:
|
|
def test_parse_capability_map(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "business-architecture" / "02-capability-map.csv")
|
|
assert "capabilities" in result
|
|
caps = result["capabilities"]
|
|
assert len(caps) > 0
|
|
cap = caps[0]
|
|
assert cap.capability_id.startswith("CAP-")
|
|
assert cap.name # should have a name
|
|
assert isinstance(cap.related_value_flows, list)
|
|
|
|
def test_capability_related_value_flows_split(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "business-architecture" / "02-capability-map.csv")
|
|
caps = result["capabilities"]
|
|
# CAP-PROGRESS-DESIGN has "VF-02 VF-03" which should be split
|
|
progress_cap = [c for c in caps if c.capability_id == "CAP-PROGRESS-DESIGN"]
|
|
assert len(progress_cap) == 1
|
|
assert progress_cap[0].related_value_flows == ["VF-02", "VF-03"]
|
|
|
|
|
|
class TestCsvParserModules:
|
|
def test_parse_modules(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "application-architecture" / "02-modules.csv")
|
|
assert "modules" in result
|
|
mods = result["modules"]
|
|
assert len(mods) > 0
|
|
mod = mods[0]
|
|
assert mod.module_id.startswith("MOD-")
|
|
assert isinstance(mod.depends_on, list)
|
|
assert isinstance(mod.capabilities, list)
|
|
|
|
def test_module_list_fields(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "application-architecture" / "02-modules.csv")
|
|
mods = result["modules"]
|
|
scanner = [m for m in mods if m.module_id == "MOD-SCANNER"]
|
|
assert len(scanner) == 1
|
|
assert "MOD-DESIGN" in scanner[0].depends_on
|
|
assert len(scanner[0].capabilities) > 0
|
|
|
|
|
|
class TestCsvParserTraceability:
|
|
def test_parse_traceability(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "traceability.csv")
|
|
assert "traceability_links" in result
|
|
links = result["traceability_links"]
|
|
assert len(links) > 0
|
|
link = links[0]
|
|
assert link.trace_id.startswith("TR-")
|
|
assert isinstance(link.entity_ids, list)
|
|
assert isinstance(link.value_flow_ids, list)
|
|
|
|
def test_traceability_space_split(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "traceability.csv")
|
|
links = result["traceability_links"]
|
|
# TR-04 has many entity_ids space-separated
|
|
tr04 = [l for l in links if l.trace_id == "TR-04"]
|
|
assert len(tr04) == 1
|
|
assert len(tr04[0].entity_ids) > 5
|
|
|
|
|
|
class TestCsvParserOtherTypes:
|
|
def test_parse_value_flows(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "business-architecture" / "03-value-flows.csv")
|
|
assert "value_flows" in result
|
|
assert len(result["value_flows"]) > 0
|
|
|
|
def test_parse_user_journeys(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "business-architecture" / "04-user-journeys.csv")
|
|
assert "user_journeys" in result
|
|
assert len(result["user_journeys"]) > 0
|
|
|
|
def test_parse_integrations(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "application-architecture" / "03-integrations.csv")
|
|
assert "integrations" in result
|
|
assert len(result["integrations"]) > 0
|
|
|
|
def test_parse_external_systems(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "application-architecture" / "01-external-systems.csv")
|
|
assert "external_systems" in result
|
|
assert len(result["external_systems"]) > 0
|
|
|
|
def test_parse_entities(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "data-architecture" / "01-entities.csv")
|
|
assert "entities" in result
|
|
assert len(result["entities"]) > 0
|
|
|
|
def test_parse_data_flows(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "data-architecture" / "02-data-flows.csv")
|
|
assert "data_flows" in result
|
|
assert len(result["data_flows"]) > 0
|
|
|
|
def test_parse_data_security(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "data-architecture" / "03-data-security.csv")
|
|
assert "data_securities" in result
|
|
assert len(result["data_securities"]) > 0
|
|
|
|
def test_parse_tech_selections(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "technology-architecture" / "00-technology-selection.csv")
|
|
assert "tech_selections" in result
|
|
assert len(result["tech_selections"]) > 0
|
|
|
|
def test_parse_runtime_components(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "technology-architecture" / "01-runtime-components.csv")
|
|
assert "runtime_components" in result
|
|
assert len(result["runtime_components"]) > 0
|
|
|
|
def test_parse_environments(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "technology-architecture" / "02-environments.csv")
|
|
assert "environments" in result
|
|
assert len(result["environments"]) > 0
|
|
|
|
def test_parse_change_log(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "change-log.csv")
|
|
assert "change_log_entries" in result
|
|
assert len(result["change_log_entries"]) > 0
|
|
|
|
def test_parse_shared_terminology(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "domains" / "_shared" / "01-shared-terminology.csv")
|
|
assert "shared_terms" in result
|
|
terms = result["shared_terms"]
|
|
assert len(terms) > 0
|
|
# Check used_by_domains is a list (space-split)
|
|
assert isinstance(terms[0].used_by_domains, list)
|
|
assert len(terms[0].used_by_domains) > 0
|
|
|
|
def test_parse_ubiquitous_language(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "domains" / "design" / "02-ubiquitous-language.csv")
|
|
assert "ubiquitous_terms" in result
|
|
assert len(result["ubiquitous_terms"]) > 0
|
|
|
|
def test_parse_scenarios(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "domains" / "design" / "03-scenarios-and-flows.csv")
|
|
assert "scenarios" in result
|
|
assert len(result["scenarios"]) > 0
|
|
|
|
def test_parse_domain_modules(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "domains" / "design" / "04-domain-modules.csv")
|
|
assert "domain_modules" in result
|
|
assert len(result["domain_modules"]) > 0
|
|
|
|
def test_parse_domain_entities(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "domains" / "design" / "05-domain-entities.csv")
|
|
assert "domain_entities" in result
|
|
assert len(result["domain_entities"]) > 0
|
|
|
|
def test_parse_codebase_alignment(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "application-architecture" / "06-codebase-alignment.csv")
|
|
assert "codebase_alignments" in result
|
|
assert len(result["codebase_alignments"]) > 0
|
|
|
|
def test_parse_codebase_mapping(self, csv_parser):
|
|
result = csv_parser.parse(DESIGN_DIR / "domains" / "design" / "07-codebase-mapping.csv")
|
|
assert "codebase_alignments" in result
|
|
assert len(result["codebase_alignments"]) > 0
|
|
|
|
|
|
class TestCsvParserUnknown:
|
|
def test_unknown_csv_returns_empty(self, csv_parser, tmp_path):
|
|
unknown = tmp_path / "unknown-file.csv"
|
|
unknown.write_text("col1,col2\nval1,val2\n")
|
|
result = csv_parser.parse(unknown)
|
|
assert result == {}
|
|
|
|
def test_nonexistent_file_returns_empty(self, csv_parser):
|
|
result = csv_parser.parse(Path("/nonexistent/file.csv"))
|
|
assert result == {}
|
|
|
|
|
|
# ── MD Parser Tests ──
|
|
|
|
from app.modules.scanner.infrastructure.parsers.md_parser import MdParser
|
|
|
|
|
|
@pytest.fixture
|
|
def md_parser():
|
|
return MdParser()
|
|
|
|
|
|
class TestMdParserScopeAndGoals:
|
|
def test_parse_scope_and_goals(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "business-architecture" / "01-scope-and-goals.md")
|
|
assert "design_documents" in result
|
|
docs = result["design_documents"]
|
|
assert len(docs) == 1
|
|
assert docs[0].doc_id == "DOC-BA-001"
|
|
assert docs[0].title == "范围与目标"
|
|
assert isinstance(docs[0].owners, list)
|
|
assert isinstance(docs[0].downstream, list)
|
|
assert "scope_and_goals" in result
|
|
sag = result["scope_and_goals"]
|
|
assert len(sag) == 1
|
|
assert sag[0].doc_id == "DOC-BA-001"
|
|
|
|
|
|
class TestMdParserDomainOverview:
|
|
def test_parse_domain_overview(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "domains" / "design" / "01-domain-overview.md")
|
|
# domain-overview.md has no frontmatter in this repo, so it produces no DesignDocument
|
|
# If it has no frontmatter, it returns empty
|
|
# Check: this file does not have frontmatter
|
|
content = (DESIGN_DIR / "domains" / "design" / "01-domain-overview.md").read_text()
|
|
if content.startswith("---"):
|
|
assert "design_documents" in result
|
|
assert "domains" in result
|
|
assert result["domains"][0].domain_name == "design"
|
|
else:
|
|
# No frontmatter, so empty result and Domain produced from filename
|
|
assert result == {} or "domains" in result
|
|
|
|
|
|
class TestMdParserSystemContext:
|
|
def test_parse_system_context(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "application-architecture" / "01-system-context.md")
|
|
assert "design_documents" in result
|
|
assert "system_context" in result
|
|
sc = result["system_context"]
|
|
assert len(sc) == 1
|
|
assert sc[0].doc_id == "DOC-AA-001"
|
|
assert sc[0].title == "系统上下文"
|
|
assert len(sc[0].content) > 0
|
|
|
|
|
|
class TestMdParserAdrTemplate:
|
|
def test_adr_template_no_adr_entity(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "adr" / "ADR-000-template.md")
|
|
# ADR-000-template has no frontmatter, so empty
|
|
content = (DESIGN_DIR / "adr" / "ADR-000-template.md").read_text()
|
|
if not content.startswith("---"):
|
|
assert result == {}
|
|
else:
|
|
# If it has frontmatter, should NOT produce ADR (it's a template)
|
|
assert "adrs" not in result
|
|
|
|
|
|
class TestMdParserNoFrontmatter:
|
|
def test_no_frontmatter_returns_empty(self, md_parser, tmp_path):
|
|
md = tmp_path / "test.md"
|
|
md.write_text("# Just a heading\n\nSome content.\n")
|
|
result = md_parser.parse(md)
|
|
assert result == {}
|
|
|
|
def test_nonexistent_md_returns_empty(self, md_parser):
|
|
result = md_parser.parse(Path("/nonexistent/file.md"))
|
|
assert result == {}
|
|
|
|
|
|
class TestMdParserSolutionLayering:
|
|
def test_parse_solution_layering(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "application-architecture" / "02b-solution-layering.md")
|
|
assert "design_documents" in result
|
|
assert "solution_layer" in result
|
|
sl = result["solution_layer"]
|
|
assert len(sl) == 1
|
|
assert sl[0].doc_id == "DOC-AA-003"
|
|
|
|
|
|
class TestMdParserModuleBoundary:
|
|
def test_parse_module_boundary(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "application-architecture" / "07-module-boundary-rules.md")
|
|
assert "design_documents" in result
|
|
assert "module_boundary_rule" in result
|
|
|
|
|
|
class TestMdParserRuntimeTopology:
|
|
def test_parse_runtime_topology(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "technology-architecture" / "01-runtime-topology.md")
|
|
assert "design_documents" in result
|
|
assert "runtime_topology" in result
|
|
|
|
|
|
class TestMdParserOperationalBaseline:
|
|
def test_parse_operational_baseline(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "technology-architecture" / "03-operational-baseline.md")
|
|
assert "design_documents" in result
|
|
assert "operational_baseline" in result
|
|
|
|
|
|
class TestMdParserReleasePlan:
|
|
def test_parse_release_plan(self, md_parser):
|
|
result = md_parser.parse(DESIGN_DIR / "technology-architecture" / "04-release-and-rollback.md")
|
|
assert "design_documents" in result
|
|
assert "release_plan" in result
|
|
|
|
|
|
# ── YAML Parser Tests ──
|
|
|
|
from app.modules.scanner.infrastructure.parsers.yaml_parser import YamlParser
|
|
|
|
|
|
@pytest.fixture
|
|
def yaml_parser():
|
|
return YamlParser()
|
|
|
|
|
|
class TestYamlParser:
|
|
def test_load_openapi_yaml(self, yaml_parser):
|
|
data = yaml_parser.load(
|
|
DESIGN_DIR / "application-architecture" / "04-api-contracts.openapi.yaml"
|
|
)
|
|
assert data is not None
|
|
assert "openapi" in data
|
|
assert "paths" in data
|
|
|
|
def test_load_nonexistent_returns_none(self, yaml_parser):
|
|
result = yaml_parser.load(Path("/nonexistent/file.yaml"))
|
|
assert result is None
|
|
|
|
def test_load_plain_yaml(self, yaml_parser, tmp_path):
|
|
f = tmp_path / "test.yaml"
|
|
f.write_text("key: value\nlist:\n - one\n - two\n")
|
|
data = yaml_parser.load(f)
|
|
assert data == {"key": "value", "list": ["one", "two"]}
|
|
|
|
|
|
# ── OpenAPI Parser Tests ──
|
|
|
|
from app.modules.scanner.infrastructure.parsers.openapi_parser import OpenapiParser
|
|
|
|
|
|
@pytest.fixture
|
|
def openapi_parser():
|
|
return OpenapiParser()
|
|
|
|
|
|
class TestOpenapiParser:
|
|
def test_parse_api_contracts(self, openapi_parser):
|
|
result = openapi_parser.parse(
|
|
DESIGN_DIR / "application-architecture" / "04-api-contracts.openapi.yaml"
|
|
)
|
|
assert "api_contracts" in result
|
|
contracts = result["api_contracts"]
|
|
assert len(contracts) > 0
|
|
# Check that contracts have correct fields
|
|
contract = contracts[0]
|
|
assert contract.path.startswith("/")
|
|
assert contract.method in ("GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD")
|
|
assert contract.doc_id.startswith("API-")
|
|
|
|
def test_parse_health_endpoint(self, openapi_parser):
|
|
result = openapi_parser.parse(
|
|
DESIGN_DIR / "application-architecture" / "04-api-contracts.openapi.yaml"
|
|
)
|
|
contracts = result["api_contracts"]
|
|
health = [c for c in contracts if c.path == "/api/health"]
|
|
assert len(health) == 1
|
|
assert health[0].method == "GET"
|
|
assert health[0].operation_id == "healthCheck"
|
|
|
|
def test_parse_nonexistent_returns_empty(self, openapi_parser):
|
|
result = openapi_parser.parse(Path("/nonexistent/file.yaml"))
|
|
assert result == {}
|
|
|
|
def test_parse_non_openapi_yaml_returns_empty(self, openapi_parser, tmp_path):
|
|
f = tmp_path / "not-openapi.yaml"
|
|
f.write_text("key: value\n")
|
|
result = openapi_parser.parse(f)
|
|
assert result == {}
|