From 51bd0fffa59131b12f7324f7b8fa36584590c2ff Mon Sep 17 00:00:00 2001 From: OpenCode Date: Fri, 10 Jul 2026 11:51:01 +0700 Subject: [PATCH] feat: Phase I I4 adapter stub -- OpenAMP output to external tool adapter with ADS- prefix, 10 target tools, dry-lab enforcement, 63 tests (#912) --- src/openamp_foundry/interop/__init__.py | 0 src/openamp_foundry/interop/adapter_stub.py | 187 ++++++++++ tests/interop/test_adapter_stub.py | 364 ++++++++++++++++++++ tests/test_test_count_regression.py | 47 ++- 4 files changed, 574 insertions(+), 24 deletions(-) create mode 100644 src/openamp_foundry/interop/__init__.py create mode 100644 src/openamp_foundry/interop/adapter_stub.py create mode 100644 tests/interop/test_adapter_stub.py diff --git a/src/openamp_foundry/interop/__init__.py b/src/openamp_foundry/interop/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/openamp_foundry/interop/adapter_stub.py b/src/openamp_foundry/interop/adapter_stub.py new file mode 100644 index 00000000..22028ebd --- /dev/null +++ b/src/openamp_foundry/interop/adapter_stub.py @@ -0,0 +1,187 @@ +"""Adapter stub: OpenAMP output → external tool format.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +ADAPTER_STUB_ID_PREFIX = "ADS-" + +VALID_TARGET_TOOLS: frozenset = frozenset( + { + "ampscanner", + "campr3", + "dbaasp", + "dramp", + "hmmer", + "iamp2l", + "modlamp", + "peptide_ranker", + "propedia", + "rosetta_fold", + } +) + +VALID_OUTPUT_FORMATS: frozenset = frozenset( + {"fasta", "csv", "json", "tsv", "pdb_sequence_list", "sdf"} +) + +VALID_ADAPTER_STATUSES: frozenset = frozenset( + {"stub", "partial", "complete", "deprecated"} +) + +TOY_ID_PREFIXES: frozenset = frozenset( + {"TOY-", "MOCK-", "EXAMPLE-", "DEMO-", "TEST-"} +) + +STUB_DISCLAIMER = ( + "This is a stub adapter. Outputs are computational nominees only. " + "No biological activity is implied. Human review required before wet-lab use." +) + + +@dataclass +class AdapterField: + source_field: str + target_field: str + transform: str + is_required: bool = True + notes: str = "" + + +@dataclass +class AdapterStubRecord: + adapter_id: str + target_tool: str + output_format: str + adapter_status: str + field_mappings: list = field(default_factory=list) + dry_lab_only: bool = True + disclaimer: str = STUB_DISCLAIMER + version: str = "0.1.0" + implementation_notes: str = "" + + +@dataclass +class AdapterValidationResult: + is_valid: bool + violations: list = field(default_factory=list) + + +def build_adapter_field( + source_field: str, + target_field: str, + transform: str = "identity", + is_required: bool = True, + notes: str = "", +) -> AdapterField: + return AdapterField( + source_field=source_field, + target_field=target_field, + transform=transform, + is_required=is_required, + notes=notes, + ) + + +def build_adapter_stub_record( + adapter_id: str, + target_tool: str, + output_format: str, + adapter_status: str = "stub", + field_mappings: list | None = None, + dry_lab_only: bool = True, + disclaimer: str = STUB_DISCLAIMER, + version: str = "0.1.0", + implementation_notes: str = "", +) -> AdapterStubRecord: + return AdapterStubRecord( + adapter_id=adapter_id, + target_tool=target_tool, + output_format=output_format, + adapter_status=adapter_status, + field_mappings=list(field_mappings) if field_mappings else [], + dry_lab_only=dry_lab_only, + disclaimer=disclaimer, + version=version, + implementation_notes=implementation_notes or f"Stub adapter for {target_tool}", + ) + + +def validate_adapter_stub_record(record: AdapterStubRecord) -> AdapterValidationResult: + violations: list[str] = [] + + if not record.adapter_id.startswith(ADAPTER_STUB_ID_PREFIX): + violations.append( + f"adapter_id must start with '{ADAPTER_STUB_ID_PREFIX}', got '{record.adapter_id}'" + ) + + if record.target_tool not in VALID_TARGET_TOOLS: + violations.append( + f"target_tool '{record.target_tool}' not in VALID_TARGET_TOOLS" + ) + + if record.output_format not in VALID_OUTPUT_FORMATS: + violations.append( + f"output_format '{record.output_format}' not in VALID_OUTPUT_FORMATS" + ) + + if record.adapter_status not in VALID_ADAPTER_STATUSES: + violations.append( + f"adapter_status '{record.adapter_status}' not in VALID_ADAPTER_STATUSES" + ) + + if not record.dry_lab_only: + violations.append( + "dry_lab_only must be True; wet-lab adapter configuration requires human review" + ) + + if not record.disclaimer: + violations.append("disclaimer must not be empty") + + if "biological activity" not in record.disclaimer.lower() and "biological" not in record.disclaimer.lower(): + violations.append( + "disclaimer must mention that no biological activity is implied" + ) + + if not record.version: + violations.append("version must not be empty") + + if not record.implementation_notes: + violations.append("implementation_notes must not be empty") + + for i, fm in enumerate(record.field_mappings): + if not fm.source_field: + violations.append(f"field_mappings[{i}]: source_field must not be empty") + if not fm.target_field: + violations.append(f"field_mappings[{i}]: target_field must not be empty") + if not fm.transform: + violations.append(f"field_mappings[{i}]: transform must not be empty") + + return AdapterValidationResult( + is_valid=len(violations) == 0, + violations=violations, + ) + + +def format_adapter_stub_record(record: AdapterStubRecord) -> str: + lines: list[str] = [] + lines.append(f"Adapter: {record.adapter_id}") + lines.append(f"Target tool: {record.target_tool}") + lines.append(f"Output format: {record.output_format}") + lines.append(f"Status: {record.adapter_status}") + lines.append(f"Version: {record.version}") + lines.append(f"Dry-lab only: {record.dry_lab_only}") + lines.append(f"Disclaimer: {record.disclaimer}") + lines.append(f"Notes: {record.implementation_notes}") + if record.field_mappings: + lines.append(f"Field mappings ({len(record.field_mappings)}):") + for fm in record.field_mappings: + req = "required" if fm.is_required else "optional" + lines.append( + f" {fm.source_field} -> {fm.target_field} [{fm.transform}] ({req})" + ) + if fm.notes: + lines.append(f" note: {fm.notes}") + else: + lines.append("Field mappings: none defined (stub)") + return "\n".join(lines) + "\n" diff --git a/tests/interop/test_adapter_stub.py b/tests/interop/test_adapter_stub.py new file mode 100644 index 00000000..f748d594 --- /dev/null +++ b/tests/interop/test_adapter_stub.py @@ -0,0 +1,364 @@ +"""Tests for src/openamp_foundry/interop/adapter_stub.py (63 tests).""" + +import pytest +from openamp_foundry.interop.adapter_stub import ( + ADAPTER_STUB_ID_PREFIX, + VALID_TARGET_TOOLS, + VALID_OUTPUT_FORMATS, + VALID_ADAPTER_STATUSES, + TOY_ID_PREFIXES, + STUB_DISCLAIMER, + AdapterField, + AdapterStubRecord, + AdapterValidationResult, + build_adapter_field, + build_adapter_stub_record, + validate_adapter_stub_record, + format_adapter_stub_record, +) + + +def _valid_record(): + return build_adapter_stub_record( + adapter_id="ADS-001", + target_tool="ampscanner", + output_format="fasta", + adapter_status="stub", + dry_lab_only=True, + disclaimer=STUB_DISCLAIMER, + version="0.1.0", + implementation_notes="Stub for AMPScanner v2 integration", + ) + + +class TestConstants: + def test_prefix(self): + assert ADAPTER_STUB_ID_PREFIX == "ADS-" + + def test_valid_target_tools_is_frozenset(self): + assert isinstance(VALID_TARGET_TOOLS, frozenset) + + def test_ampscanner_in_tools(self): + assert "ampscanner" in VALID_TARGET_TOOLS + + def test_modlamp_in_tools(self): + assert "modlamp" in VALID_TARGET_TOOLS + + def test_hmmer_in_tools(self): + assert "hmmer" in VALID_TARGET_TOOLS + + def test_valid_output_formats_is_frozenset(self): + assert isinstance(VALID_OUTPUT_FORMATS, frozenset) + + def test_fasta_in_formats(self): + assert "fasta" in VALID_OUTPUT_FORMATS + + def test_csv_in_formats(self): + assert "csv" in VALID_OUTPUT_FORMATS + + def test_json_in_formats(self): + assert "json" in VALID_OUTPUT_FORMATS + + def test_valid_adapter_statuses_is_frozenset(self): + assert isinstance(VALID_ADAPTER_STATUSES, frozenset) + + def test_stub_in_statuses(self): + assert "stub" in VALID_ADAPTER_STATUSES + + def test_complete_in_statuses(self): + assert "complete" in VALID_ADAPTER_STATUSES + + def test_deprecated_in_statuses(self): + assert "deprecated" in VALID_ADAPTER_STATUSES + + def test_toy_id_prefixes_is_frozenset(self): + assert isinstance(TOY_ID_PREFIXES, frozenset) + + def test_toy_prefix_in_prefixes(self): + assert "TOY-" in TOY_ID_PREFIXES + + def test_stub_disclaimer_not_empty(self): + assert STUB_DISCLAIMER != "" + + def test_stub_disclaimer_mentions_biological(self): + assert "biological" in STUB_DISCLAIMER.lower() + + def test_stub_disclaimer_mentions_human_review(self): + assert "human review" in STUB_DISCLAIMER.lower() + + +class TestBuildAdapterField: + def test_returns_adapter_field(self): + f = build_adapter_field("candidate_id", "sequence_id") + assert isinstance(f, AdapterField) + + def test_source_field_stored(self): + f = build_adapter_field("candidate_id", "sequence_id") + assert f.source_field == "candidate_id" + + def test_target_field_stored(self): + f = build_adapter_field("candidate_id", "sequence_id") + assert f.target_field == "sequence_id" + + def test_transform_default_identity(self): + f = build_adapter_field("candidate_id", "sequence_id") + assert f.transform == "identity" + + def test_transform_custom(self): + f = build_adapter_field("sequence", "aa_sequence", transform="uppercase") + assert f.transform == "uppercase" + + def test_is_required_default_true(self): + f = build_adapter_field("candidate_id", "sequence_id") + assert f.is_required is True + + def test_is_required_false(self): + f = build_adapter_field("description", "notes", is_required=False) + assert f.is_required is False + + def test_notes_stored(self): + f = build_adapter_field("candidate_id", "id", notes="rename only") + assert f.notes == "rename only" + + +class TestBuildAdapterStubRecord: + def test_returns_adapter_stub_record(self): + record = _valid_record() + assert isinstance(record, AdapterStubRecord) + + def test_adapter_id_stored(self): + record = _valid_record() + assert record.adapter_id == "ADS-001" + + def test_target_tool_stored(self): + record = _valid_record() + assert record.target_tool == "ampscanner" + + def test_output_format_stored(self): + record = _valid_record() + assert record.output_format == "fasta" + + def test_adapter_status_stored(self): + record = _valid_record() + assert record.adapter_status == "stub" + + def test_dry_lab_only_stored(self): + record = _valid_record() + assert record.dry_lab_only is True + + def test_disclaimer_stored(self): + record = _valid_record() + assert record.disclaimer == STUB_DISCLAIMER + + def test_version_stored(self): + record = _valid_record() + assert record.version == "0.1.0" + + def test_implementation_notes_auto_when_empty(self): + record = build_adapter_stub_record( + adapter_id="ADS-001", + target_tool="ampscanner", + output_format="fasta", + ) + assert record.implementation_notes != "" + + def test_field_mappings_empty_by_default(self): + record = _valid_record() + assert record.field_mappings == [] + + def test_field_mappings_stored(self): + fm = build_adapter_field("candidate_id", "seq_id") + record = build_adapter_stub_record( + adapter_id="ADS-001", + target_tool="ampscanner", + output_format="fasta", + field_mappings=[fm], + ) + assert len(record.field_mappings) == 1 + + +class TestValidateAdapterStubRecord: + def test_valid_record_passes(self): + result = validate_adapter_stub_record(_valid_record()) + assert result.is_valid + + def test_returns_validation_result(self): + result = validate_adapter_stub_record(_valid_record()) + assert isinstance(result, AdapterValidationResult) + + def test_no_violations_on_valid(self): + result = validate_adapter_stub_record(_valid_record()) + assert result.violations == [] + + def test_wrong_id_prefix_fails(self): + record = _valid_record() + record.adapter_id = "BAD-001" + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("ADS-" in v for v in result.violations) + + def test_invalid_target_tool_fails(self): + record = _valid_record() + record.target_tool = "unknown_tool" + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("target_tool" in v for v in result.violations) + + def test_invalid_output_format_fails(self): + record = _valid_record() + record.output_format = "docx" + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("output_format" in v for v in result.violations) + + def test_invalid_adapter_status_fails(self): + record = _valid_record() + record.adapter_status = "unknown" + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("adapter_status" in v for v in result.violations) + + def test_dry_lab_false_fails(self): + record = _valid_record() + record.dry_lab_only = False + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("dry_lab_only" in v for v in result.violations) + + def test_empty_disclaimer_fails(self): + record = _valid_record() + record.disclaimer = "" + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("disclaimer" in v for v in result.violations) + + def test_disclaimer_missing_biological_fails(self): + record = _valid_record() + record.disclaimer = "Stub adapter. Human review required." + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("biological" in v for v in result.violations) + + def test_empty_version_fails(self): + record = _valid_record() + record.version = "" + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("version" in v for v in result.violations) + + def test_empty_implementation_notes_fails(self): + record = _valid_record() + record.implementation_notes = "" + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("implementation_notes" in v for v in result.violations) + + def test_field_mapping_empty_source_fails(self): + fm = build_adapter_field("", "target") + record = build_adapter_stub_record( + adapter_id="ADS-001", + target_tool="ampscanner", + output_format="fasta", + field_mappings=[fm], + implementation_notes="test", + ) + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("source_field" in v for v in result.violations) + + def test_field_mapping_empty_target_fails(self): + fm = build_adapter_field("source", "") + record = build_adapter_stub_record( + adapter_id="ADS-001", + target_tool="ampscanner", + output_format="fasta", + field_mappings=[fm], + implementation_notes="test", + ) + result = validate_adapter_stub_record(record) + assert not result.is_valid + assert any("target_field" in v for v in result.violations) + + def test_multiple_violations_accumulated(self): + record = _valid_record() + record.adapter_id = "BAD-001" + record.target_tool = "unknown" + result = validate_adapter_stub_record(record) + assert len(result.violations) >= 2 + + def test_all_valid_target_tools_accepted(self): + for tool in VALID_TARGET_TOOLS: + record = build_adapter_stub_record( + adapter_id="ADS-001", + target_tool=tool, + output_format="fasta", + implementation_notes=f"stub for {tool}", + ) + result = validate_adapter_stub_record(record) + assert result.is_valid, f"Tool {tool} failed: {result.violations}" + + def test_all_valid_formats_accepted(self): + for fmt in VALID_OUTPUT_FORMATS: + record = build_adapter_stub_record( + adapter_id="ADS-001", + target_tool="ampscanner", + output_format=fmt, + implementation_notes="stub", + ) + result = validate_adapter_stub_record(record) + assert result.is_valid, f"Format {fmt} failed: {result.violations}" + + +class TestFormatAdapterStubRecord: + def test_returns_string(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert isinstance(result, str) + + def test_contains_adapter_id(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert "ADS-001" in result + + def test_contains_target_tool(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert "ampscanner" in result + + def test_contains_output_format(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert "fasta" in result + + def test_contains_status(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert "stub" in result + + def test_contains_disclaimer(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert "biological" in result.lower() + + def test_ends_with_newline(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert result.endswith("\n") + + def test_field_mappings_shown(self): + fm = build_adapter_field("candidate_id", "seq_id", transform="identity") + record = build_adapter_stub_record( + adapter_id="ADS-001", + target_tool="ampscanner", + output_format="fasta", + field_mappings=[fm], + implementation_notes="test", + ) + result = format_adapter_stub_record(record) + assert "candidate_id" in result + assert "seq_id" in result + + def test_no_mappings_shows_stub_message(self): + record = _valid_record() + result = format_adapter_stub_record(record) + assert "none defined" in result or "stub" in result diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index fc6f6a62..6526828e 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -1,35 +1,34 @@ -"""Regression guard: total collected test count must not drop below BASELINE.""" -import subprocess -import sys +"""Regression test: total test count must stay within 5% of BASELINE.""" +import subprocess +import math -BASELINE = 7906 -TOLERANCE = 0.05 +BASELINE = 8158 def test_test_count_regression(): result = subprocess.run( - [sys.executable, "-m", "pytest", "--collect-only", "-q", "--no-header"], + ["python", "-m", "pytest", "--collect-only", "-q", "--no-header"], capture_output=True, text=True, - cwd=".", ) - output = result.stdout + result.stderr - count = 0 - for line in output.splitlines(): - if "test session starts" in line or "selected" in line or "no tests ran" in line: - continue - if line.strip().endswith(("PASSED", "FAILED", "ERROR", "SKIPPED")): - count += 1 - for line in output.splitlines(): - if "selected" in line and "test" in line: - import re - m = re.search(r"(\d+) test", line) - if m: - count = int(m.group(1)) + lines = result.stdout.strip().splitlines() + count_line = next( + (l for l in reversed(lines) if "test" in l and ("selected" in l or "error" in l)), + None, + ) + if count_line is None: + for l in reversed(lines): + if l.strip() and l[0].isdigit(): + count_line = l break - min_allowed = int(BASELINE * (1 - TOLERANCE)) - assert count >= min_allowed, ( - f"Test count dropped from {BASELINE} to {count} " - f"(threshold: {min_allowed}). Update BASELINE if intentional." + assert count_line is not None, f"Could not find test count line. Output:\n{result.stdout}\n" + import re + m = re.search(r"(\d+)", count_line) + assert m is not None, f"No number in count line: {count_line}" + actual = int(m.group(1)) + tolerance = math.ceil(BASELINE * 0.05) + assert abs(actual - BASELINE) <= tolerance, ( + f"Test count {actual} deviates from BASELINE {BASELINE} by more than {tolerance}. " + f"Update BASELINE in this file if tests were intentionally added or removed." )