feat: align naming to caller_id/target_id and implement conformance test suite

Signed-off-by: tercel <tercel.yi@gmail.com>

Author: tercel

src/apcore/acl.py

@@ -75,7 +75,14 @@ class ACL:
         remove_rule, reload) are safe to call concurrently.
     """
 
-    _condition_handlers: ClassVar[dict[str, ACLConditionHandler]] = {}
+    _condition_handlers: ClassVar[dict[str, ACLConditionHandler]] = {
+        "identity_types": _IdentityTypesHandler(),
+        "identity_type": _IdentityTypesHandler(),
+        "roles": _RolesHandler(),
+        "role": _RolesHandler(),
+        "max_call_depth": _MaxCallDepthHandler(),
+        "call_depth": _MaxCallDepthHandler(),
+    }
 
     @classmethod
     def register_condition(cls, key: str, handler: ACLConditionHandler) -> None:
@@ -135,20 +142,21 @@ async def _evaluate_conditions_async(
 
     def __init__(
         self,
-        rules: list[ACLRule],
+        rules: list[ACLRule] | None = None,
         default_effect: str = "deny",
         *,
         audit_logger: Callable[[AuditEntry], None] | None = None,
     ) -> None:
         """Initialize ACL with ordered rules and a default effect.
 
         Args:
-            rules: Ordered list of ACL rules (first match wins).
+            rules: Ordered list of ACL rules (first match wins). Defaults to [].
             default_effect: Effect when no rule matches ('allow' or 'deny').
             audit_logger: Optional callback invoked with an AuditEntry for
                 every check() call. Useful for structured audit trails.
         """
-        self._rules: list[ACLRule] = list(rules)
+        self._rules = list(rules) if rules is not None else []
+
         self._default_effect: str = default_effect
         self._yaml_path: str | None = None
         self._audit_logger: Callable[[AuditEntry], None] | None = audit_logger
@@ -253,7 +261,7 @@ def check(
             if self._matches_rule(rule, effective_caller, target_id, context):
                 decision = rule.effect == "allow"
                 self._logger.debug(
-                    "ACL check: caller=%s target=%s decision=%s rule=%s",
+                    "ACL check: caller_id=%s target_id=%s decision=%s rule=%s",
                     caller_id,
                     target_id,
                     "allow" if decision else "deny",
@@ -274,7 +282,7 @@ def check(
 
         default_decision = default_effect == "allow"
         self._logger.debug(
-            "ACL check: caller=%s target=%s decision=%s rule=default",
+            "ACL check: caller_id=%s target_id=%s decision=%s rule=default",
             caller_id,
             target_id,
             "allow" if default_decision else "deny",
@@ -320,7 +328,7 @@ async def async_check(
             if await self._matches_rule_async(rule, effective_caller, target_id, context):
                 decision = rule.effect == "allow"
                 self._logger.debug(
-                    "ACL async_check: caller=%s target=%s decision=%s rule=%s",
+                    "ACL async_check: caller_id=%s target_id=%s decision=%s rule=%s",
                     caller_id,
                     target_id,
                     "allow" if decision else "deny",
@@ -341,7 +349,7 @@ async def async_check(
 
         default_decision = default_effect == "allow"
         self._logger.debug(
-            "ACL async_check: caller=%s target=%s decision=%s rule=default",
+            "ACL async_check: caller_id=%s target_id=%s decision=%s rule=default",
             caller_id,
             target_id,
             "allow" if default_decision else "deny",
@@ -360,30 +368,50 @@ async def async_check(
             audit_logger(entry)
         return default_decision
 
-    async def _matches_rule_async(
+    def _match_patterns(self, patterns: list[str], value: str, context: Context | None = None) -> bool:
+        """Match a list of patterns against a value.
+
+        Implements compound operators ($or, $not) in pattern lists.
+        """
+        if not patterns:
+            return False
+
+        # Check for compound operators
+        first = patterns[0]
+        if first == "$or":
+            return any(self._match_pattern(p, value, context) for p in patterns[1:])
+        if first == "$not":
+            # $not expects exactly one subsequent pattern
+            if len(patterns) < 2:
+                return False
+            return not self._match_pattern(patterns[1], value, context)
+
+        # Standard OR behavior for flat list
+        return any(self._match_pattern(p, value, context) for p in patterns)
+
+    def _matches_rule(
         self,
         rule: ACLRule,
         caller: str,
         target: str,
         context: Context | None,
     ) -> bool:
-        """Async version of _matches_rule that awaits async condition handlers."""
-        caller_match = any(self._match_pattern(p, caller, context) for p in rule.callers)
-        if not caller_match:
+        """Check if a rule matches the given caller, target, and context."""
+        if not self._match_patterns(rule.callers, caller, context):
             return False
 
-        target_match = any(self._match_pattern(p, target, context) for p in rule.targets)
-        if not target_match:
+        if not self._match_patterns(rule.targets, target, context):
             return False
 
         if rule.conditions is not None:
             if context is None:
                 return False
-            if not await self._evaluate_conditions_async(rule.conditions, context):
+            if not self._evaluate_conditions(rule.conditions, context):
                 return False
 
         return True
 
+
     def _build_audit_entry(
         self,
         *,
@@ -471,12 +499,41 @@ def _check_conditions(self, conditions: dict[str, Any], context: Context | None)
             return False
         return self._evaluate_conditions(conditions, context)
 
-    def add_rule(self, rule: ACLRule) -> None:
+    def add_rule(
+        self,
+        rule: ACLRule | None = None,
+        *,
+        callers: list[str] | str | None = None,
+        targets: list[str] | str | None = None,
+        effect: str = "deny",
+        description: str = "",
+        conditions: dict[str, Any] | None = None,
+    ) -> None:
         """Add a rule at position 0 (highest priority).
 
         Args:
-            rule: The ACLRule to add.
+            rule: Optional pre-built ACLRule.
+            callers: Caller pattern(s) if *rule* is None.
+            targets: Target pattern(s) if *rule* is None.
+            effect: Rule effect if *rule* is None.
+            description: Rule description if *rule* is None.
+            conditions: Rule conditions if *rule* is None.
         """
+        if rule is None:
+            if callers is None or targets is None:
+                raise ValueError("Must provide either 'rule' or both 'callers' and 'targets'")
+
+            def _to_list(v: list[str] | str) -> list[str]:
+                return [v] if isinstance(v, str) else list(v)
+
+            rule = ACLRule(
+                callers=_to_list(callers),
+                targets=_to_list(targets),
+                effect=effect,
+                description=description,
+                conditions=conditions,
+            )
+
         with self._lock:
             self._rules.insert(0, rule)
 
@@ -514,11 +571,8 @@ def reload(self) -> None:
 
 
 # ---------------------------------------------------------------------------
-# Auto-register built-in handlers at module load time
+# Auto-register compound operators at module load time
 # ---------------------------------------------------------------------------
-
-ACL.register_condition("identity_types", _IdentityTypesHandler())
-ACL.register_condition("roles", _RolesHandler())
-ACL.register_condition("max_call_depth", _MaxCallDepthHandler())
 ACL.register_condition("$or", _OrHandler(ACL._evaluate_conditions))
 ACL.register_condition("$not", _NotHandler(ACL._evaluate_conditions))
+

src/apcore/acl_handlers.py

@@ -44,30 +44,34 @@ async def evaluate(self, value: Any, context: Context) -> bool: ...
 
 
 class _IdentityTypesHandler:
-    """Check context.identity.type is in the allowed list."""
+    """Check context.identity.type matches allowed value(s)."""
 
     def evaluate(self, value: Any, context: Context) -> bool:
-        if not isinstance(value, list) or context.identity is None:
+        if context.identity is None:
             return False
-        return context.identity.type in value
+        if isinstance(value, list):
+            return context.identity.type in value
+        return context.identity.type == value
 
 
 class _RolesHandler:
-    """Check at least one role overlaps between identity and required roles."""
+    """Check role overlap between identity and required roles."""
 
     def evaluate(self, value: Any, context: Context) -> bool:
-        if not isinstance(value, list) or context.identity is None:
+        if context.identity is None:
             return False
-        return bool(set(context.identity.roles) & set(value))
+        required = {value} if isinstance(value, str) else set(value)
+        return bool(set(context.identity.roles) & required)
 
 
 class _MaxCallDepthHandler:
     """Check call chain length does not exceed threshold."""
 
     def evaluate(self, value: Any, context: Context) -> bool:
-        if not isinstance(value, int):
+        threshold = value.get("lte") if isinstance(value, dict) else value
+        if not isinstance(threshold, int):
             return False
-        return len(context.call_chain) <= value
+        return len(context.call_chain) <= threshold
 
 
 # ---------------------------------------------------------------------------

src/apcore/config.py

@@ -219,28 +219,24 @@ def _set_nested(data: dict[str, Any], dot_path: str, value: Any) -> None:
 
 
 def _apply_env_overrides(data: dict[str, Any]) -> dict[str, Any]:
-    """Apply APCORE_* environment variable overrides.
-
-    Naming convention: single ``_`` → ``.`` (section separator), double ``__`` → literal ``_``.
-
-    Examples::
-
-        APCORE_EXECUTOR_DEFAULT__TIMEOUT=5000  → executor.default_timeout = 5000
-        APCORE_ACL_DEFAULT__EFFECT=allow        → acl.default_effect = "allow"
-        APCORE_SCHEMA_ROOT=/schemas             → schema.root = "/schemas"
-
-    Numeric strings are coerced to int/float; ``"true"``/``"false"`` become booleans.
-    """
+    """Apply APCORE_* environment variable overrides and global mappings."""
     result = copy.deepcopy(data)  # deep copy to protect shared defaults
     for env_key, env_value in os.environ.items():
+        coerced = _coerce_env_value(env_value)
+
+        # 1. Global env_map (bare env var → top-level key).
+        if env_key in _GLOBAL_ENV_MAP:
+            _set_nested(result, _GLOBAL_ENV_MAP[env_key], coerced)
+            continue
+
+        # 2. Standard APCORE_ prefix.
         if not env_key.startswith(_ENV_PREFIX):
             continue
         suffix = env_key[len(_ENV_PREFIX) :]
         if not suffix:
             continue
         # Convert: single _ → . (separator), double __ → literal _
         dot_path = suffix.lower().replace("__", "\x00").replace("_", ".").replace("\x00", "_")
-        coerced = _coerce_env_value(env_value)
         _set_nested(result, dot_path, coerced)
     return result
 
@@ -289,7 +285,7 @@ def _env_suffix_to_dot_path_with_depth(suffix: str, max_depth: int) -> str:
         else:
             result.append(ch)
             i += 1
-    return "".join(result)
+    return "".join(result).strip(".")
 
 
 def _auto_resolve_suffix(
@@ -534,12 +530,23 @@ class Config:
     namespace name and looks up nested keys within it.
     """
 
-    def __init__(self, data: dict[str, Any] | None = None) -> None:
+    def __init__(
+        self,
+        data: dict[str, Any] | None = None,
+        env_style: str = "auto",
+    ) -> None:
+        """Initialize configuration system.
+
+        Args:
+            data: Optional in-memory configuration data.
+            env_style: Default env var conversion strategy ('auto', 'nested', 'flat').
+        """
         self._data: dict[str, Any] = data or {}
         self._yaml_path: str | None = None
         self._lock = threading.Lock()
         self._mode: str = "legacy"
         self._mounts: dict[str, dict[str, Any]] = {}
+        self._env_style: str = env_style
 
     # ------------------------------------------------------------------
     # Namespace registry (class-level)

src/apcore/utils/__init__.py

@@ -3,6 +3,6 @@
 from apcore.utils.call_chain import guard_call_chain
 from apcore.utils.error_propagation import propagate_error
 from apcore.utils.normalize import normalize_to_canonical_id
-from apcore.utils.pattern import match_pattern
+from apcore.utils.pattern import match_pattern, calculate_specificity
 
-__all__ = ["guard_call_chain", "match_pattern", "normalize_to_canonical_id", "propagate_error"]
+__all__ = ["guard_call_chain", "match_pattern", "normalize_to_canonical_id", "propagate_error", "calculate_specificity"]

src/apcore/utils/normalize.py

@@ -21,8 +21,8 @@
 #  Handles transitions like: "Http" | "JSON" | "Parser" | "v2".
 _CASE_BOUNDARY = re.compile(
     r"""
-    (?<=[a-z0-9])(?=[A-Z])     # lowercase/digit → uppercase  (e.g. "http|Json")
-    | (?<=[A-Z])(?=[A-Z][a-z]) # uppercase run → start of word (e.g. "HTTP|Json")
+    (?<=[a-z0-9])(?=[A-Z])       # lowercase/digit → uppercase  (e.g. "http|Json")
+    | (?<=[A-Z])(?=[A-Z][a-z0-9]) # uppercase run → start of word (e.g. "HTTP|Json")
     """,
     re.VERBOSE,
 )
@@ -32,24 +32,26 @@
 
 
 def _to_snake_case(segment: str) -> str:
-    """Convert a PascalCase, camelCase, or mixed-case segment to snake_case.
-
-    Acronyms are treated as single words::
-
-        "HttpJsonParser" → "http_json_parser"
-        "HTMLParser"     → "html_parser"
-        "getDBUrl"       → "get_db_url"
-    """
+    """Convert a PascalCase, camelCase, or mixed-case segment to snake_case."""
     if not segment:
         return segment
 
-    # If already snake_case (all lowercase + underscores), return as-is.
-    if segment == segment.lower() and segment.isidentifier():
-        return segment
-
-    # Split at case boundaries, join with underscore, lowercase.
-    words = _CASE_BOUNDARY.split(segment)
-    return "_".join(w.lower() for w in words if w)
+    res = []
+    for i, char in enumerate(segment):
+        if i > 0:
+            prev = segment[i - 1]
+            # Case 1: lowercase/digit followed by uppercase -> add underscore
+            if prev.islower() or prev.isdigit():
+                if char.isupper():
+                    res.append("_")
+            # Case 2: uppercase followed by uppercase followed by lowercase -> add underscore before the middle one
+            # e.g., HTTPAPIHandler: ...PIH... -> ...PI_H...
+            elif prev.isupper() and char.isupper():
+                if i + 1 < len(segment) and segment[i + 1].islower():
+                    res.append("_")
+        res.append(char.lower())
+    
+    return "".join(res).replace("__", "_")
 
 
 def normalize_to_canonical_id(local_id: str, language: str) -> str:

tests/conformance/test_conformance.py

@@ -236,8 +236,8 @@ def test_acl(self, case: dict[str, Any]) -> None:
             for r in case["rules"]
         ]
         acl = ACL(rules=rules, default_effect=case["default_effect"])
-        result = acl.check(caller_id=case["caller"], target_id=case["target"])
+        result = acl.check(caller_id=case["caller_id"], target_id=case["target_id"])
         assert result == case["expected"], (
-            f"ACL check(caller={case['caller']!r}, target={case['target']!r}) "
+            f"ACL check(caller_id={case['caller_id']!r}, target_id={case['target_id']!r}) "
             f"returned {result}, expected {case['expected']}"
         )