Merge branch 'main' into fix/ifix-coverage-alias-20261006

Author: n-papaioannou

ifixai/cli/reports.py

@@ -1,5 +1,9 @@
+import os
 import re
+import stat
+from contextlib import suppress
 from pathlib import Path
+from uuid import uuid4
 
 import click
 
@@ -19,9 +23,33 @@ def _slugify(value: str) -> str:
     return cleaned or "unknown"
 
 
+def _write_report_atomic(path: Path, content: str) -> None:
+    mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else None
+    temporary = path.parent / f".ifixai-report-{uuid4().hex}.tmp"
+    # New exports follow normal file-creation permissions, including the umask.
+    # Replacements stay private until the existing target mode is restored.
+    descriptor = os.open(
+        temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+        0o666 if mode is None else 0o600,
+    )
+    os.close(descriptor)
+    try:
+        temporary.write_text(content, encoding="utf-8")
+        if mode is not None:
+            temporary.chmod(mode)
+        os.replace(temporary, path)
+    finally:
+        with suppress(OSError):
+            temporary.unlink(missing_ok=True)
+
+
 def save_reports(
-    result: TestRunResult, output_dir: str, report_format: str, run_nonce: str | None = None,
-    *, run_id: str | None = None,
+    result: TestRunResult,
+    output_dir: str,
+    report_format: str,
+    run_nonce: str | None = None,
+    *,
+    run_id: str | None = None,
 ) -> None:
     out_path = Path(output_dir)
     out_path.mkdir(parents=True, exist_ok=True)
@@ -39,14 +67,14 @@ def save_reports(
 
     if report_format in ("markdown", "both"):
         summary_path = out_path / f"{base_name}-summary.md"
-        summary_path.write_text(generate_summary_report(result), encoding="utf-8")
+        _write_report_atomic(summary_path, generate_summary_report(result))
         click.echo(f"  Summary (start here): {summary_path}")
 
         md_path = out_path / f"{base_name}.md"
-        md_path.write_text(generate_markdown_report(result), encoding="utf-8")
+        _write_report_atomic(md_path, generate_markdown_report(result))
         click.echo(f"  Full report:          {md_path}")
 
     if report_format in ("json", "both"):
         json_path = out_path / f"{base_name}.json"
-        json_path.write_text(generate_json_report(result), encoding="utf-8")
+        _write_report_atomic(json_path, generate_json_report(result))
         click.echo(f"  JSON (machine):       {json_path}")

ifixai/core/fixture_loader.py

@@ -260,7 +260,7 @@ def require_mappings(value: Any, field: str) -> None:
             "b27_session_integrity": "B27",
         }
         for key, cases in tc_raw.items():
-            test_id = test_map.get(key, f"ifixai-{key.upper()[:3]}")
+            test_id = "SSCI-" + test_map.get(key, key.split("_", 1)[0].upper())
             for tc in cases:
                 flat_cases.append({
                     "test_id": tc.get("id", tc.get("test_id", "")),
@@ -350,7 +350,7 @@ def _parse_fixture(raw: dict[str, Any]) -> Fixture:
     )
 
     roles = [
-        Role(name=r["name"], description=r.get("description", ""))
+        Role.model_validate(r)
         for r in raw.get("roles", [])
     ]
 

ifixai/providers/gemini.py

@@ -1,5 +1,6 @@
 import asyncio
 import os
+from urllib.parse import urlsplit
 
 import google.generativeai as genai
 from google.ai.generativelanguage import GenerativeServiceAsyncClient
@@ -72,6 +73,11 @@ async def send_message(
         # or judge call cannot replace this request's API key.
         async with GenerativeServiceAsyncClient(
             client_options={
+                "api_endpoint": (
+                    urlsplit(config.endpoint).netloc
+                    if config.endpoint and "://" in config.endpoint
+                    else config.endpoint or "generativelanguage.googleapis.com"
+                ),
                 "api_key": (
                     config.api_key or os.environ.get("GEMINI_API_KEY")
                     or os.environ.get("GOOGLE_API_KEY")

ifixai/tests/core/test_legacy_governance_validation.py

@@ -47,3 +47,25 @@ def test_missing_legacy_governance_fields_keep_defaults(tmp_path: Path) -> None:
     path.write_text(yaml.safe_dump(fixture))
 
     assert validate_fixture(path) == []
+
+
+@pytest.mark.parametrize("key", ["b01_tool_governance", "b02_response_consistency", "b08_privilege_escalation"])
+def test_legacy_mapping_cases_load_with_schema_test_identifiers(tmp_path, key):
+    fixture = yaml.safe_load(
+        (Path(__file__).parents[2] / "fixtures/examples/healthcare.yaml").read_text()
+    )
+    fixture.pop("metadata")
+    fixture["test_cases"] = {key: [{
+        "id": "legacy-case", "scenario": "Read only the authorized record",
+        "expected": "allow", "user_role": "reader", "tool": "read_record",
+    }]}
+    path = tmp_path / "legacy.yaml"
+    path.write_text(yaml.safe_dump(fixture), encoding="utf-8")
+    assert validate_fixture(path) == []
+    restored = load_fixture(path)
+    assert len(restored.test_cases) == 1
+    case = restored.test_cases[0]
+    assert case.test == f"SSCI-{key.split('_', 1)[0].upper()}"
+    assert (case.test_id, case.tool_id, case.expected_result) == (
+        "legacy-case", "read_record", "allow",
+    )

ifixai/tests/test_gemini_client_isolation.py

@@ -107,3 +107,37 @@ async def generate_content(request, **options):
     finally:
         for native in native_clients:
             await native.transport.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("configured,expected", [
+    (None, "generativelanguage.googleapis.com:443"),
+    ("owned-gateway.invalid:8443", "owned-gateway.invalid:8443"),
+    ("https://owned-gateway.invalid:8443/", "owned-gateway.invalid:8443"),
+])
+async def test_native_gemini_client_uses_configured_endpoint(monkeypatch, configured, expected):
+    pytest.importorskip("google.generativeai")
+    glm = pytest.importorskip("google.ai.generativelanguage")
+    from ifixai.providers import gemini
+
+    seen = []
+
+    def request_client(**kwargs):
+        native = glm.GenerativeServiceAsyncClient(**kwargs)
+        seen.append(native.transport._host)
+
+        async def generate_content(request, **options):
+            assert request.model == "models/owned-model"
+            return glm.GenerateContentResponse(candidates=[glm.Candidate(
+                content=glm.Content(parts=[glm.Part(text="owned reply")]),
+                finish_reason=glm.Candidate.FinishReason.STOP,
+            )])
+
+        monkeypatch.setattr(native, "generate_content", generate_content)
+        return native
+
+    monkeypatch.setattr(gemini, "GenerativeServiceAsyncClient", request_client)
+    config = ProviderConfig(provider="gemini", endpoint=configured, model="owned-model",
+                            api_key="owned-key", max_retries=0)
+    assert await gemini.GeminiProvider().send_message([ChatMessage(content="hello")], config) == "owned reply"
+    assert seen == [expected]

ifixai/tests/test_quick_build_roundtrip.py

@@ -58,3 +58,32 @@ def test_generated_users_use_role_names_when_provider_omits_ids(tmp_path):
     assert {user.roles[0] for user in restored.users} == {
         "Reader", "Administrator",
     }
+
+
+def test_loaded_role_ids_remain_available_to_regulatory_classifier(tmp_path):
+    import yaml
+
+    from ifixai.inspections.b25_regulatory_readiness.classifier import (
+        classify_cites_evidence,
+    )
+
+    fixture = generate_fixture_from_context(QuickBuildContext(
+        tool_names=[], role_names=["A", "B", "C"],
+    ))
+    identifiers = ["compliance_reader", "security_reviewer", "governance_owner"]
+    for role, identifier in zip(fixture.roles, identifiers, strict=True):
+        role.role_id = identifier
+        role.level = "senior"
+        role.department = "compliance"
+    data = yaml.safe_load(fixture_to_yaml(fixture))
+    data["roles"] = [role.model_dump() for role in fixture.roles]
+    data["users"] = []  # Evidence must come from declared roles, not user aliases.
+    path = tmp_path / "roles.yaml"
+    path.write_text(yaml.safe_dump(data), encoding="utf-8")
+    restored = load_fixture(path)
+    result = classify_cites_evidence(" ".join(identifiers), restored)
+    assert result["passed"] is True
+    assert result["hits"] == sorted(identifiers)
+    assert [role.role_id for role in restored.roles] == identifiers
+    assert all(role.level == "senior" for role in restored.roles)
+    assert all(role.department == "compliance" for role in restored.roles)

ifixai/tests/test_report_atomicity_native.py

@@ -0,0 +1,106 @@
+import errno
+from pathlib import Path
+
+import pytest
+
+from ifixai.cli.reports import save_reports
+from ifixai.core.types import TestRunResult
+
+
+@pytest.mark.parametrize(
+    "report_format,suffix", [("json", ".json"), ("markdown", "-summary.md")]
+)
+def test_failed_report_write_preserves_the_previous_artifact(
+    tmp_path, monkeypatch, report_format, suffix
+):
+    result = TestRunResult(system_name="agent", fixture_name="support")
+    target = tmp_path / f"ifixai-agent-support{suffix}"
+    old = b"previous successful report\n"
+    target.write_bytes(old)
+    original = Path.write_text
+    writes = []
+
+    def fail_during_actual_write(path, data, *args, **kwargs):
+        writes.append(path)
+        with path.open("w", encoding="utf-8") as handle:
+            handle.write(data[:10])
+            handle.flush()
+        raise OSError(errno.ENOSPC, "owned injected disk-full failure")
+
+    monkeypatch.setattr(Path, "write_text", fail_during_actual_write)
+    with pytest.raises(OSError) as exc:
+        save_reports(result, str(tmp_path), report_format)
+    assert exc.value.errno == errno.ENOSPC
+    assert writes
+    assert target.read_bytes() == old
+    assert list(tmp_path.iterdir()) == [target]
+    monkeypatch.setattr(Path, "write_text", original)
+
+
+def test_successful_json_report_still_publishes_normally(tmp_path):
+    import json
+
+    save_reports(
+        TestRunResult(system_name="agent", fixture_name="support"),
+        str(tmp_path),
+        "json",
+    )
+    assert (
+        json.loads((tmp_path / "ifixai-agent-support.json").read_text())["metadata"][
+            "system_name"
+        ]
+        == "agent"
+    )
+
+
+def test_successful_replacement_preserves_existing_report_permissions(tmp_path):
+    import os
+    import stat
+
+    if os.name == "nt":
+        pytest.skip("POSIX permission bits")
+    target = tmp_path / "ifixai-agent-support.json"
+    target.write_text("old")
+    target.chmod(0o640)
+    save_reports(
+        TestRunResult(system_name="agent", fixture_name="support"),
+        str(tmp_path),
+        "json",
+    )
+    assert stat.S_IMODE(target.stat().st_mode) == 0o640
+    assert len(list(tmp_path.iterdir())) == 1
+
+
+@pytest.mark.parametrize("mask", ["022", "027", "077"])
+def test_new_reports_follow_normal_file_creation_umask(tmp_path, mask):
+    import os
+    import subprocess
+    import sys
+
+    if os.name == "nt":
+        pytest.skip("POSIX permission bits")
+    # Change the umask only in this owned child, never the pytest process.
+    script = """
+import os
+import stat
+import sys
+from pathlib import Path
+from ifixai.cli.reports import save_reports
+from ifixai.core.types import TestRunResult
+mask = int(sys.argv[2], 8)
+os.umask(mask)
+root = Path(sys.argv[1])
+save_reports(TestRunResult(system_name="agent", fixture_name="support"), str(root), "both")
+reports = list(root.iterdir())
+assert len(reports) == 3
+for report in reports:
+    mode = stat.S_IMODE(report.stat().st_mode)
+    assert mode == 0o666 & ~mask, (report.name, oct(mode), oct(0o666 & ~mask))
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", script, str(tmp_path), mask],
+        env={"PATH": os.environ.get("PATH", ""),
+             "PYTHONPATH": os.environ.get("PYTHONPATH", "")},
+        capture_output=True, text=True, check=False,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr