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:
co-authored by
Claude Opus 4.8
parent
cbe3ba66c6
commit
af350fe07e
@@ -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.
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user