"""Step definitions for docker_compose_analyzer_coverage.feature. These steps target specific uncovered lines in docker_compose_analyzer.py: - Lines 142-145, 147: Size-guard branch (content exceeds _MAX_COMPOSE_BYTES) - Line 164: YAML parses to a non-dict (scalar/list/None) → return [] - Line 194: services key present but not a dict → return triples (deploy only) - Lines 255-263: Exception during service processing → log warning + continue """ import logging from unittest.mock import patch from behave import given, then, when from cleveragents.domain.models.acms.docker_compose_analyzer import ( _MAX_COMPOSE_BYTES, DockerComposeAnalyzer, ) # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the docker compose analyzer module is imported") def step_module_imported(context): """Ensure the module is importable and create a fresh analyzer.""" context.analyzer = DockerComposeAnalyzer() assert context.analyzer is not None # --------------------------------------------------------------------------- # Scenario: Content exceeding the 1 MiB size limit (lines 142-145, 147) # --------------------------------------------------------------------------- @given("I have YAML content that exceeds the 1 MiB byte limit") def step_create_oversized_content(context): """Build YAML content larger than _MAX_COMPOSE_BYTES.""" # Pad with a YAML comment so it stays valid but oversized. padding = ( "# " + "x" * (_MAX_COMPOSE_BYTES + 100) + "\nservices:\n web:\n image: nginx\n" ) context.oversized_content = padding assert len(context.oversized_content.encode("utf-8")) > _MAX_COMPOSE_BYTES @when("I analyze the oversized content") def step_analyze_oversized(context): """Run the analyzer on oversized content and capture the log.""" context.log_records = [] class _CapturingHandler(logging.Handler): def __init__(self, records_list): super().__init__() self._records = records_list def emit(self, record): self._records.append(record) handler = _CapturingHandler(context.log_records) logger = logging.getLogger( "cleveragents.domain.models.acms.docker_compose_analyzer" ) # Re-enable in case Alembic's fileConfig() disabled it logger.disabled = False logger.addHandler(handler) logger.setLevel(logging.DEBUG) context.oversized_result = context.analyzer.analyze( context.oversized_content, "infra/docker-compose.yml" ) logger.removeHandler(handler) @then("the analyzer should return an empty list") def step_verify_empty_list(context): """The oversized input must produce no triples.""" assert context.oversized_result == [] @then("a size-limit warning should have been logged") def step_verify_size_warning(context): """Verify the warning about the byte limit was emitted.""" warnings = [r for r in context.log_records if r.levelno == logging.WARNING] assert any("exceeds" in r.getMessage() for r in warnings), ( f"Expected a size-limit warning, got: {[r.getMessage() for r in warnings]}" ) # --------------------------------------------------------------------------- # Scenario: YAML that parses to a non-dict scalar (line 164) # --------------------------------------------------------------------------- @given("I have YAML content that parses to a plain scalar string") def step_create_scalar_yaml(context): """Create YAML content that safe_load returns as a bare string.""" context.scalar_yaml = "just a plain string\n" @when("I analyze the non-dict YAML content") def step_analyze_scalar(context): """Run the analyzer on scalar-only YAML.""" context.scalar_result = context.analyzer.analyze( context.scalar_yaml, "infra/not-compose.yaml" ) @then("the analyzer should return an empty list for non-dict input") def step_verify_empty_for_non_dict(context): """Scalar YAML must produce no triples.""" assert context.scalar_result == [] # --------------------------------------------------------------------------- # Scenario: services key present but not a dict (line 194) # --------------------------------------------------------------------------- @given("I have compose YAML where services is a scalar instead of a mapping") def step_create_services_scalar_yaml(context): """Create YAML where ``services`` is a string, not a mapping.""" context.non_dict_services_yaml = "services: not-a-mapping\n" @when("I analyze the compose YAML with non-dict services") def step_analyze_non_dict_services(context): """Run the analyzer on YAML with a non-dict ``services`` value.""" context.non_dict_services_result = context.analyzer.analyze( context.non_dict_services_yaml, "infra/docker-compose.yml" ) @then("the analyzer should return only the deployment unit triples") def step_verify_deployment_only(context): """Only the DeploymentUnit declaration + label triples should be present.""" triples = context.non_dict_services_result # Exactly 2 triples: rdf:type DeploymentUnit + rdfs:label assert len(triples) == 2, f"Expected 2 deployment triples, got {len(triples)}" predicates = {t.predicate for t in triples} assert "rdf:type" in predicates assert "rdfs:label" in predicates type_triple = next(t for t in triples if t.predicate == "rdf:type") assert type_triple.object_uri == "uko-infra:DeploymentUnit" # --------------------------------------------------------------------------- # Scenario: Exception during service processing (lines 255-263) # --------------------------------------------------------------------------- @given("I have valid compose YAML with two services") def step_create_two_service_yaml(context): """Create compose YAML with two services: 'web' and 'db'.""" context.two_svc_yaml = ( "services:\n" " web:\n" " image: nginx\n" " ports:\n" ' - "80:80"\n' " db:\n" " image: postgres\n" ) @given("the port extraction method is patched to raise an error for one service") def step_patch_extract_ports(context): """Patch _extract_ports so it raises on the 'web' service only.""" original = DockerComposeAnalyzer._extract_ports def _exploding_ports(self, service_def, resource_uri, service_name, svc_uri): if str(service_name) == "web": raise RuntimeError("Simulated extraction failure") return original(self, service_def, resource_uri, service_name, svc_uri) context.ports_patcher = patch.object( DockerComposeAnalyzer, "_extract_ports", _exploding_ports ) context.ports_patcher.start() def cleanup(): context.ports_patcher.stop() context.add_cleanup(cleanup) @when("I analyze the compose YAML with the faulty extractor") def step_analyze_with_faulty_extractor(context): """Run the analyzer; the exception for 'web' should be caught.""" context.faulty_log_records = [] class _CapturingHandler(logging.Handler): def __init__(self, records_list): super().__init__() self._records = records_list def emit(self, record): self._records.append(record) handler = _CapturingHandler(context.faulty_log_records) logger = logging.getLogger( "cleveragents.domain.models.acms.docker_compose_analyzer" ) # Re-enable in case Alembic's fileConfig() disabled it logger.disabled = False logger.addHandler(handler) logger.setLevel(logging.DEBUG) context.faulty_result = context.analyzer.analyze( context.two_svc_yaml, "infra/docker-compose.yml" ) logger.removeHandler(handler) @then("the analyzer should return triples for the non-failing service") def step_verify_non_failing_service_triples(context): """The 'db' service should still produce triples even though 'web' failed.""" triples = context.faulty_result # We should have deployment triples + db service triples. # web service may have partial triples (declared before ports extraction). db_triples = [ t for t in triples if "db" in (t.subject_uri or "") or "db" in (t.object_uri or "") ] assert len(db_triples) > 0, f"Expected db service triples, got none in {triples}" @then("a service processing warning should have been logged") def step_verify_service_warning(context): """Verify the warning about the service processing error was emitted.""" warnings = [r for r in context.faulty_log_records if r.levelno == logging.WARNING] assert any("error processing service" in r.getMessage() for r in warnings), ( f"Expected a service error warning, got: {[r.getMessage() for r in warnings]}" )