fix(commons): canonical lowercase identity + fast test suite (green CI push)

Identity: normalize branch names to lowercase at both write paths so one
branch = one identity regardless of registry casing (registry has historically
mixed BACKUP vs devpulse, splitting the roster into DEVPULSE/devpulse rows).
- identity_ops.get_caller_branch(): _normalize_branch_name() at the single
  caller choke point (post/comment author writes + agent registration).
- db._register_branches(): lowercase on the bulk registry seed.
Verified live: post author lands lowercase, no duplicate rows; uppercase-
registry branches (backup) normalize through the caller path too.

Test suite: session-scoped template DB cloned per test (shutil.copy) + fast
PRAGMAs (journal_mode=MEMORY, synchronous=OFF) instead of re-running
schema.sql+FTS5+registry per test. 449 tests now 86s (was >120s gate timeout);
full per-test isolation preserved, initialized_db interface unchanged.

test_identity: assertions updated for the lowercase caller path; monkeypatch
targets retargeted from the commons_identity facade to identity_ops (where
get_caller_branch resolves them).

.daemon/schedule.json: disabled wake-test seed (decentralized daemon contract example).

seedgo 100% (37/37), 449 tests pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-06-15 17:45:08 -07:00
co-authored by Claude Opus 4.8
parent cbe3ba66c6
commit af350fe07e
5 changed files with 151 additions and 91 deletions
+18
View File
@@ -0,0 +1,18 @@
{
"version": 1,
"branch": "@commons",
"jobs": [
{
"id": "wake-test",
"enabled": false,
"schedule": {
"type": "interval",
"interval_minutes": 1
},
"wake": {
"fresh": true
},
"prompt": "AUTOMATED DAEMON TEST. Do ONLY this: run drone @ai_mail email @devpulse \"DAEMON TEST FIRED\" \"Wake received from @daemon scheduler. No memory updated.\" then STOP. Do NOT run startup, do NOT update memory, do NOT do anything else."
}
]
}
@@ -354,7 +354,10 @@ def _register_branches(conn: sqlite3.Connection) -> None:
branches = registry.get("branches", []) branches = registry.get("branches", [])
for branch in branches: for branch in branches:
name = branch.get("name", "") # Lowercase to keep one branch = one identity regardless of registry
# casing (registry has historically mixed BACKUP vs devpulse). Matches
# the normalization in identity_ops.get_caller_branch().
name = branch.get("name", "").lower()
if not name: if not name:
continue continue
@@ -190,6 +190,7 @@ def get_caller_branch() -> Optional[Dict[str, Any]]:
if branch_root: if branch_root:
branch_info = get_branch_info_from_registry(branch_root) branch_info = get_branch_info_from_registry(branch_root)
if branch_info: if branch_info:
_normalize_branch_name(branch_info)
_ensure_agent_registered(branch_info) _ensure_agent_registered(branch_info)
json_handler.log_operation("caller_detected", {"branch": branch_info.get("name", "unknown")}) json_handler.log_operation("caller_detected", {"branch": branch_info.get("name", "unknown")})
return branch_info return branch_info
@@ -199,6 +200,7 @@ def get_caller_branch() -> Optional[Dict[str, Any]]:
if caller_branch_name: if caller_branch_name:
branch_info = get_branch_info_by_name(caller_branch_name) branch_info = get_branch_info_by_name(caller_branch_name)
if branch_info: if branch_info:
_normalize_branch_name(branch_info)
_ensure_agent_registered(branch_info) _ensure_agent_registered(branch_info)
json_handler.log_operation( json_handler.log_operation(
"caller_detected", {"branch": branch_info.get("name", "unknown"), "via": "AIPASS_CALLER_BRANCH"} "caller_detected", {"branch": branch_info.get("name", "unknown"), "via": "AIPASS_CALLER_BRANCH"}
@@ -215,6 +217,25 @@ def get_caller_branch() -> Optional[Dict[str, Any]]:
return None return None
def _normalize_branch_name(branch_info: Dict[str, Any]) -> None:
"""
Lowercase the branch name so Commons identity is case-canonical.
The AIPASS registry has historically used mixed casing for branch names
(e.g. BACKUP vs devpulse), and both the agents roster and post authorship
mirror whatever casing the registry supplies. get_caller_branch() is the
single choke point every caller identity flows through (post/comment author
writes, agent registration), so normalizing here keeps one branch = one
identity regardless of registry casing — no DEVPULSE/devpulse splits.
Args:
branch_info: Branch dict from the registry (mutated in place).
"""
name = branch_info.get("name")
if isinstance(name, str) and name:
branch_info["name"] = name.lower()
def _ensure_agent_registered(branch_info: Dict[str, Any]) -> None: def _ensure_agent_registered(branch_info: Dict[str, Any]) -> None:
""" """
Ensure the branch is registered as an agent in The Commons database. Ensure the branch is registered as an agent in The Commons database.
+37 -23
View File
@@ -1,16 +1,9 @@
# ===================AIPASS==================== # =================== AIPass ====================
# META DATA HEADER # Name: conftest.py
# Name: conftest.py - The Commons test configuration # Description: The Commons test configuration
# Date: 2026-03-07 # Version: 1.1.0
# Version: 1.0.0 # Created: 2026-03-07
# Category: commons/tests # Modified: 2026-06-15
#
# CHANGELOG (Max 5 entries):
# - v1.0.0 (2026-03-07): Initial creation (FPLAN-0411)
#
# CODE STANDARDS:
# - Pytest fixtures for The Commons test suite
# - Uses temporary database for test isolation
# ============================================= # =============================================
""" """
@@ -21,6 +14,8 @@ and test isolation using temporary databases.
""" """
import os import os
import shutil
import sqlite3
import tempfile import tempfile
# Redirect prax logs to temp directory during tests # Redirect prax logs to temp directory during tests
@@ -29,14 +24,28 @@ if "AIPASS_TEST_LOG_DIR" not in os.environ:
os.environ["AIPASS_TEST_LOG_DIR"] = tempfile.mkdtemp(prefix="aipass_test_logs_") os.environ["AIPASS_TEST_LOG_DIR"] = tempfile.mkdtemp(prefix="aipass_test_logs_")
import pytest import logging # noqa: E402
import pytest # noqa: E402
logger = logging.getLogger(__name__)
try: try:
from aipass.prax.apps.modules.logger import system_logger as logger from aipass.prax.apps.modules.logger import system_logger as logger # noqa: E402, F811
except ImportError: except ImportError:
import logging logger.warning("[conftest] prax logger unavailable — using stdlib logging")
logger = logging.getLogger("commons.tests")
@pytest.fixture(scope="session")
def _template_db_path(tmp_path_factory):
"""Build the initialized schema+seed DB once per session."""
from aipass.commons.apps.modules.database import close_db, init_db
template = tmp_path_factory.mktemp("template") / "template_commons.db"
conn = init_db(db_path=template)
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
close_db(conn)
return template
@pytest.fixture @pytest.fixture
@@ -55,21 +64,26 @@ def tmp_db_path(tmp_path):
@pytest.fixture @pytest.fixture
def initialized_db(tmp_db_path): def initialized_db(_template_db_path, tmp_path):
""" """
Provide an initialized temporary database with schema and seed data. Provide an initialized temporary database with schema and seed data.
Creates a fresh database with all tables, default rooms, Copies from a session-scoped template instead of re-running init_db,
and room personalities. Closes the connection after the test. keeping the interface stable (yields sqlite3.Connection).
Yields: Yields:
sqlite3.Connection to the initialized test database. sqlite3.Connection to the initialized test database.
""" """
from aipass.commons.apps.handlers.database.db import init_db, close_db db_file = tmp_path / "test_commons.db"
shutil.copy2(str(_template_db_path), str(db_file))
conn = init_db(db_path=tmp_db_path) conn = sqlite3.connect(str(db_file), timeout=30)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA journal_mode = MEMORY")
conn.execute("PRAGMA synchronous = OFF")
yield conn yield conn
close_db(conn) conn.close()
@pytest.fixture @pytest.fixture
+71 -67
View File
@@ -1,17 +1,9 @@
# ===================AIPASS==================== # =================== AIPass ====================
# META DATA HEADER # Name: test_identity.py
# Name: test_identity.py - Identity Module Unit Tests # Description: Unit tests for identity module and identity_ops handler
# Date: 2026-03-24 # Version: 1.1.0
# Version: 1.0.0 # Created: 2026-03-24
# Category: commons/tests # Modified: 2026-06-15
#
# CHANGELOG (Max 5 entries):
# - v1.0.0 (2026-03-24): Initial creation — 14 unit tests
#
# CODE STANDARDS:
# - Pytest function style (no unittest classes)
# - Mock heavy deps (prax logger, database)
# - Tests extract_mentions, find_branch_root, resolve_display_name
# ============================================= # =============================================
""" """
@@ -21,14 +13,15 @@ Tests extract_mentions (pure regex), find_branch_root (filesystem walk),
resolve_display_name, and DB-backed mention validation. resolve_display_name, and DB-backed mention validation.
""" """
import logging
import sqlite3 import sqlite3
import sys
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import pytest import pytest
# Mock the prax logger before importing the modules under test logger = logging.getLogger(__name__)
import sys
_mock_logger = MagicMock() _mock_logger = MagicMock()
_mock_logger_module = MagicMock() _mock_logger_module = MagicMock()
@@ -37,28 +30,22 @@ _mock_logger_module.system_logger = _mock_logger
try: try:
from aipass.prax.apps.modules.logger import system_logger # noqa: F401 from aipass.prax.apps.modules.logger import system_logger # noqa: F401
except ImportError: except ImportError:
logger.warning("[test_identity] prax unavailable — injecting mock logger")
sys.modules.setdefault("aipass.prax", MagicMock()) sys.modules.setdefault("aipass.prax", MagicMock())
sys.modules.setdefault("aipass.prax.apps", MagicMock()) sys.modules.setdefault("aipass.prax.apps", MagicMock())
sys.modules.setdefault("aipass.prax.apps.modules", MagicMock()) sys.modules.setdefault("aipass.prax.apps.modules", MagicMock())
sys.modules.setdefault("aipass.prax.apps.modules.logger", _mock_logger_module) sys.modules.setdefault("aipass.prax.apps.modules.logger", _mock_logger_module)
# Mock CLI console too — commons_identity imports it
try: try:
from aipass.cli.apps.modules import console # noqa: F401 from aipass.cli.apps.modules import console # noqa: F401
except ImportError: except ImportError:
logger.warning("[test_identity] cli unavailable — injecting mock console")
_mock_cli = MagicMock() _mock_cli = MagicMock()
sys.modules.setdefault("aipass.cli", _mock_cli) sys.modules.setdefault("aipass.cli", _mock_cli)
sys.modules.setdefault("aipass.cli.apps", MagicMock()) sys.modules.setdefault("aipass.cli.apps", MagicMock())
sys.modules.setdefault("aipass.cli.apps.modules", MagicMock()) sys.modules.setdefault("aipass.cli.apps.modules", MagicMock())
from aipass.commons.apps.modules.commons_identity import extract_mentions from aipass.commons.apps.modules import commons_identity as _id_mod # noqa: E402
from aipass.commons.apps.handlers.identity.identity_ops import (
find_branch_root,
get_branch_info_by_name,
get_caller_branch,
resolve_display_name,
)
import aipass.commons.apps.handlers.identity.identity_ops as identity_ops_mod
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -91,13 +78,13 @@ def _patch_db_for_mentions(initialized_db: sqlite3.Connection):
def test_extract_mentions_empty_string(initialized_db: sqlite3.Connection): def test_extract_mentions_empty_string(initialized_db: sqlite3.Connection):
"""Empty string returns empty list.""" """Empty string returns empty list."""
result = extract_mentions("") result = _id_mod.extract_mentions("")
assert result == [] assert result == []
def test_extract_mentions_no_mentions(initialized_db: sqlite3.Connection): def test_extract_mentions_no_mentions(initialized_db: sqlite3.Connection):
"""Text without @mentions returns empty list.""" """Text without @mentions returns empty list."""
result = extract_mentions("Hello world, no mentions here") result = _id_mod.extract_mentions("Hello world, no mentions here")
assert result == [] assert result == []
@@ -109,7 +96,7 @@ def test_extract_mentions_single(initialized_db: sqlite3.Connection):
) )
initialized_db.commit() initialized_db.commit()
result = extract_mentions("Hey @drone check this out") result = _id_mod.extract_mentions("Hey @drone check this out")
assert result == ["drone"] assert result == ["drone"]
@@ -122,13 +109,13 @@ def test_extract_mentions_multiple(initialized_db: sqlite3.Connection):
) )
initialized_db.commit() initialized_db.commit()
result = extract_mentions("@flow and @seed please review") result = _id_mod.extract_mentions("@flow and @seed please review")
assert result == ["flow", "seed"] assert result == ["flow", "seed"]
def test_extract_mentions_unregistered_filtered(initialized_db: sqlite3.Connection): def test_extract_mentions_unregistered_filtered(initialized_db: sqlite3.Connection):
"""Mentions of agents not in the DB are filtered out.""" """Mentions of agents not in the DB are filtered out."""
result = extract_mentions("@nonexistent_branch please help") result = _id_mod.extract_mentions("@nonexistent_branch please help")
assert result == [] assert result == []
@@ -140,7 +127,7 @@ def test_extract_mentions_case_insensitive(initialized_db: sqlite3.Connection):
) )
initialized_db.commit() initialized_db.commit()
result = extract_mentions("Hey @PRAX look at this") result = _id_mod.extract_mentions("Hey @PRAX look at this")
assert result == ["prax"] assert result == ["prax"]
@@ -152,7 +139,7 @@ def test_extract_mentions_with_underscores(initialized_db: sqlite3.Connection):
) )
initialized_db.commit() initialized_db.commit()
result = extract_mentions("Asking @ai_mail for analysis") result = _id_mod.extract_mentions("Asking @ai_mail for analysis")
assert result == ["ai_mail"] assert result == ["ai_mail"]
@@ -170,7 +157,7 @@ def test_find_branch_root_with_trinity(tmp_path: Path):
sub = tmp_path / "apps" / "handlers" sub = tmp_path / "apps" / "handlers"
sub.mkdir(parents=True) sub.mkdir(parents=True)
result = find_branch_root(sub) result = _id_mod.find_branch_root(sub)
assert result is not None assert result is not None
assert result == tmp_path.resolve() assert result == tmp_path.resolve()
@@ -180,7 +167,7 @@ def test_find_branch_root_no_trinity(tmp_path: Path):
sub = tmp_path / "deep" / "nested" / "dir" sub = tmp_path / "deep" / "nested" / "dir"
sub.mkdir(parents=True) sub.mkdir(parents=True)
result = find_branch_root(sub) result = _id_mod.find_branch_root(sub)
assert result is None assert result is None
@@ -190,7 +177,7 @@ def test_find_branch_root_at_start(tmp_path: Path):
trinity_dir.mkdir() trinity_dir.mkdir()
(trinity_dir / "passport.json").write_text("{}", encoding="utf-8") (trinity_dir / "passport.json").write_text("{}", encoding="utf-8")
result = find_branch_root(tmp_path) result = _id_mod.find_branch_root(tmp_path)
assert result is not None assert result is not None
assert result == tmp_path.resolve() assert result == tmp_path.resolve()
@@ -202,30 +189,29 @@ def test_find_branch_root_at_start(tmp_path: Path):
def test_resolve_display_name_no_alias(monkeypatch: pytest.MonkeyPatch): def test_resolve_display_name_no_alias(monkeypatch: pytest.MonkeyPatch):
"""Falls back to branch_name when no alias is cached.""" """Falls back to branch_name when no alias is cached."""
# Reset the alias cache to a known state monkeypatch.setattr("aipass.commons.apps.handlers.identity.identity_ops._alias_cache", {})
monkeypatch.setattr(identity_ops_mod, "_alias_cache", {}) result = _id_mod.resolve_display_name("UNKNOWN_BRANCH")
result = resolve_display_name("UNKNOWN_BRANCH")
assert result == "UNKNOWN_BRANCH" assert result == "UNKNOWN_BRANCH"
def test_resolve_display_name_with_alias(monkeypatch: pytest.MonkeyPatch): def test_resolve_display_name_with_alias(monkeypatch: pytest.MonkeyPatch):
"""Returns 'Alias (SYSTEM_NAME)' format when alias exists.""" """Returns 'Alias (SYSTEM_NAME)' format when alias exists."""
monkeypatch.setattr(identity_ops_mod, "_alias_cache", {"TEAM_1": "Alpha Team"}) monkeypatch.setattr("aipass.commons.apps.handlers.identity.identity_ops._alias_cache", {"TEAM_1": "Alpha Team"})
result = resolve_display_name("TEAM_1") result = _id_mod.resolve_display_name("TEAM_1")
assert result == "Alpha Team (TEAM_1)" assert result == "Alpha Team (TEAM_1)"
def test_resolve_display_name_compact(monkeypatch: pytest.MonkeyPatch): def test_resolve_display_name_compact(monkeypatch: pytest.MonkeyPatch):
"""Compact mode returns alias only, no parenthesized system name.""" """Compact mode returns alias only, no parenthesized system name."""
monkeypatch.setattr(identity_ops_mod, "_alias_cache", {"TEAM_1": "Alpha Team"}) monkeypatch.setattr("aipass.commons.apps.handlers.identity.identity_ops._alias_cache", {"TEAM_1": "Alpha Team"})
result = resolve_display_name("TEAM_1", compact=True) result = _id_mod.resolve_display_name("TEAM_1", compact=True)
assert result == "Alpha Team" assert result == "Alpha Team"
def test_resolve_display_name_compact_no_alias(monkeypatch: pytest.MonkeyPatch): def test_resolve_display_name_compact_no_alias(monkeypatch: pytest.MonkeyPatch):
"""Compact mode without alias still falls back to branch_name.""" """Compact mode without alias still falls back to branch_name."""
monkeypatch.setattr(identity_ops_mod, "_alias_cache", {}) monkeypatch.setattr("aipass.commons.apps.handlers.identity.identity_ops._alias_cache", {})
result = resolve_display_name("RAW_NAME", compact=True) result = _id_mod.resolve_display_name("RAW_NAME", compact=True)
assert result == "RAW_NAME" assert result == "RAW_NAME"
@@ -246,9 +232,12 @@ def test_get_branch_info_by_name_found(tmp_path: Path, monkeypatch: pytest.Monke
} }
reg_file = tmp_path / "AIPASS_REGISTRY.json" reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json_mod.dumps(registry), encoding="utf-8") reg_file.write_text(json_mod.dumps(registry), encoding="utf-8")
monkeypatch.setattr(identity_ops_mod, "BRANCH_REGISTRY_PATH", reg_file) monkeypatch.setattr(
"aipass.commons.apps.handlers.identity.identity_ops.BRANCH_REGISTRY_PATH",
reg_file,
)
result = get_branch_info_by_name("drone") result = _id_mod.get_branch_info_by_name("drone")
assert result is not None assert result is not None
assert result["name"] == "DRONE" assert result["name"] == "DRONE"
assert result["email"] == "@drone" assert result["email"] == "@drone"
@@ -261,9 +250,12 @@ def test_get_branch_info_by_name_case_insensitive(tmp_path: Path, monkeypatch: p
registry = {"branches": [{"name": "FLOW", "path": "src/aipass/flow"}]} registry = {"branches": [{"name": "FLOW", "path": "src/aipass/flow"}]}
reg_file = tmp_path / "AIPASS_REGISTRY.json" reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json_mod.dumps(registry), encoding="utf-8") reg_file.write_text(json_mod.dumps(registry), encoding="utf-8")
monkeypatch.setattr(identity_ops_mod, "BRANCH_REGISTRY_PATH", reg_file) monkeypatch.setattr(
"aipass.commons.apps.handlers.identity.identity_ops.BRANCH_REGISTRY_PATH",
reg_file,
)
result = get_branch_info_by_name("Flow") result = _id_mod.get_branch_info_by_name("Flow")
assert result is not None assert result is not None
assert result["name"] == "FLOW" assert result["name"] == "FLOW"
@@ -275,16 +267,22 @@ def test_get_branch_info_by_name_not_found(tmp_path: Path, monkeypatch: pytest.M
registry = {"branches": [{"name": "DRONE", "path": "src/aipass/drone"}]} registry = {"branches": [{"name": "DRONE", "path": "src/aipass/drone"}]}
reg_file = tmp_path / "AIPASS_REGISTRY.json" reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json_mod.dumps(registry), encoding="utf-8") reg_file.write_text(json_mod.dumps(registry), encoding="utf-8")
monkeypatch.setattr(identity_ops_mod, "BRANCH_REGISTRY_PATH", reg_file) monkeypatch.setattr(
"aipass.commons.apps.handlers.identity.identity_ops.BRANCH_REGISTRY_PATH",
reg_file,
)
result = get_branch_info_by_name("nonexistent") result = _id_mod.get_branch_info_by_name("nonexistent")
assert result is None assert result is None
def test_get_branch_info_by_name_missing_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): def test_get_branch_info_by_name_missing_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
"""Returns None when registry file doesn't exist.""" """Returns None when registry file doesn't exist."""
monkeypatch.setattr(identity_ops_mod, "BRANCH_REGISTRY_PATH", tmp_path / "nope.json") monkeypatch.setattr(
result = get_branch_info_by_name("DRONE") "aipass.commons.apps.handlers.identity.identity_ops.BRANCH_REGISTRY_PATH",
tmp_path / "nope.json",
)
result = _id_mod.get_branch_info_by_name("DRONE")
assert result is None assert result is None
@@ -307,17 +305,19 @@ def test_get_caller_branch_uses_caller_branch_env(
registry = {"branches": [{"name": "DRONE", "path": "src/aipass/drone", "email": "@drone"}]} registry = {"branches": [{"name": "DRONE", "path": "src/aipass/drone", "email": "@drone"}]}
reg_file = tmp_path / "AIPASS_REGISTRY.json" reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json_mod.dumps(registry), encoding="utf-8") reg_file.write_text(json_mod.dumps(registry), encoding="utf-8")
monkeypatch.setattr(identity_ops_mod, "BRANCH_REGISTRY_PATH", reg_file) monkeypatch.setattr(
"aipass.commons.apps.handlers.identity.identity_ops.BRANCH_REGISTRY_PATH",
reg_file,
)
# CWD with no .trinity/ — simulates running from project root
no_branch_dir = tmp_path / "somewhere" no_branch_dir = tmp_path / "somewhere"
no_branch_dir.mkdir() no_branch_dir.mkdir()
monkeypatch.setenv("AIPASS_CALLER_CWD", str(no_branch_dir)) monkeypatch.setenv("AIPASS_CALLER_CWD", str(no_branch_dir))
monkeypatch.setenv("AIPASS_CALLER_BRANCH", "drone") monkeypatch.setenv("AIPASS_CALLER_BRANCH", "drone")
result = get_caller_branch() result = _id_mod.get_caller_branch()
assert result is not None assert result is not None
assert result["name"] == "DRONE" assert result["name"] == "drone"
mock_register.assert_called_once() mock_register.assert_called_once()
@@ -332,33 +332,37 @@ def test_get_caller_branch_prefers_cwd_over_env(
"""CWD-based detection takes priority over AIPASS_CALLER_BRANCH.""" """CWD-based detection takes priority over AIPASS_CALLER_BRANCH."""
import json as json_mod import json as json_mod
# Set up a branch directory with .trinity/
trinity = tmp_path / ".trinity" trinity = tmp_path / ".trinity"
trinity.mkdir() trinity.mkdir()
(trinity / "passport.json").write_text("{}", encoding="utf-8") (trinity / "passport.json").write_text("{}", encoding="utf-8")
registry = { registry = {
"branches": [ "branches": [
{"name": "FLOW", "path": str(tmp_path.relative_to(tmp_path.parent.parent)), "email": "@flow"}, {
"name": "FLOW",
"path": str(tmp_path.relative_to(tmp_path.parent.parent)),
"email": "@flow",
},
] ]
} }
reg_file = tmp_path / "AIPASS_REGISTRY.json" reg_file = tmp_path / "AIPASS_REGISTRY.json"
reg_file.write_text(json_mod.dumps(registry), encoding="utf-8") reg_file.write_text(json_mod.dumps(registry), encoding="utf-8")
monkeypatch.setattr(identity_ops_mod, "BRANCH_REGISTRY_PATH", reg_file) monkeypatch.setattr(
"aipass.commons.apps.handlers.identity.identity_ops.BRANCH_REGISTRY_PATH",
reg_file,
)
# CWD is inside the branch
monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path)) monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
# Also set CALLER_BRANCH to something different — should NOT be used
monkeypatch.setenv("AIPASS_CALLER_BRANCH", "DRONE") monkeypatch.setenv("AIPASS_CALLER_BRANCH", "DRONE")
# Need to patch get_branch_info_from_registry to return for our tmp_path with patch(
with patch.object( "aipass.commons.apps.handlers.identity.identity_ops.get_branch_info_from_registry",
identity_ops_mod, "get_branch_info_from_registry", return_value={"name": "FLOW", "email": "@flow"} return_value={"name": "FLOW", "email": "@flow"},
): ):
result = get_caller_branch() result = _id_mod.get_caller_branch()
assert result is not None assert result is not None
assert result["name"] == "FLOW" # CWD-based, not DRONE from env assert result["name"] == "flow"
@patch("aipass.commons.apps.handlers.identity.identity_ops.json_handler") @patch("aipass.commons.apps.handlers.identity.identity_ops.json_handler")
@@ -373,5 +377,5 @@ def test_get_caller_branch_returns_none_when_no_detection(
monkeypatch.setenv("AIPASS_CALLER_CWD", str(no_branch_dir)) monkeypatch.setenv("AIPASS_CALLER_CWD", str(no_branch_dir))
monkeypatch.delenv("AIPASS_CALLER_BRANCH", raising=False) monkeypatch.delenv("AIPASS_CALLER_BRANCH", raising=False)
result = get_caller_branch() result = _id_mod.get_caller_branch()
assert result is None assert result is None