add storage module, decr bump to 3.12 for python version, add tests for logging and storage

Author: lilydu

packages/cards/pyproject.toml

@@ -6,7 +6,7 @@ readme = "README.md"
 authors = [
     { name = "lilydu", email = "[email protected]" }
 ]
-requires-python = ">=3.13"
+requires-python = ">=3.12"
 dependencies = [
     "coverage>=7.8.0",
     "pytest>=8.3.5",

packages/common/pyproject.toml

@@ -9,7 +9,7 @@ license = "MIT"
 authors = [
     { name = "Microsoft", email = "[email protected]" }
 ]
-requires-python = ">=3.13"
+requires-python = ">=3.12"
 dependencies = [
     "coverage>=7.8.0",
     "pytest>=8.3.5",

packages/common/src/common/logging/test_filter.py

@@ -0,0 +1,79 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+
+import logging
+from unittest.mock import MagicMock
+
+from .filter import ConsoleFilter
+
+def test_default_pattern():
+    filter = ConsoleFilter()
+    record = MagicMock(spec=logging.LogRecord)
+    record.name = "test"
+    
+    assert filter.filter(record) is True
+
+def test_exact_match():
+    filter = ConsoleFilter("test")
+    record = MagicMock(spec=logging.LogRecord)
+    record.name = "test"
+    
+    assert filter.filter(record) is True
+
+def test_wildcard_prefix():
+    filter = ConsoleFilter("test*")
+    
+    matching_record = MagicMock(spec=logging.LogRecord)
+    matching_record.name = "testLogger"
+    assert filter.filter(matching_record) is True
+    
+    non_matching_record = MagicMock(spec=logging.LogRecord)
+    non_matching_record.name = "logger"
+    assert filter.filter(non_matching_record) is False
+
+def test_wildcard_suffix():
+    filter = ConsoleFilter("*test")
+    
+    matching_record = MagicMock(spec=logging.LogRecord)
+    matching_record.name = "mytest"
+    assert filter.filter(matching_record) is True
+    
+    non_matching_record = MagicMock(spec=logging.LogRecord)
+    non_matching_record.name = "tester"
+    assert filter.filter(non_matching_record) is False
+
+def test_wildcard_middle():
+    filter = ConsoleFilter("my*test")
+    
+    matching_record = MagicMock(spec=logging.LogRecord)
+    matching_record.name = "myloggertest"
+    assert filter.filter(matching_record) is True
+    
+    non_matching_record = MagicMock(spec=logging.LogRecord)
+    non_matching_record.name = "mylogger"
+    assert filter.filter(non_matching_record) is False
+
+def test_multiple_wildcards():
+    filter = ConsoleFilter("my*log*test")
+    
+    matching_record = MagicMock(spec=logging.LogRecord)
+    matching_record.name = "myapplicationloggertest"
+    assert filter.filter(matching_record) is True
+    
+    partial_match_record = MagicMock(spec=logging.LogRecord)
+    partial_match_record.name = "mylogger"
+    assert filter.filter(partial_match_record) is False
+
+def test_case_sensitivity():
+    filter = ConsoleFilter("Test")
+    
+    upper_record = MagicMock(spec=logging.LogRecord)
+    upper_record.name = "Test"
+    assert filter.filter(upper_record) is True
+    
+    lower_record = MagicMock(spec=logging.LogRecord)
+    lower_record.name = "test"
+    assert filter.filter(lower_record) is False

packages/common/src/common/logging/test_formatter.py

@@ -0,0 +1,109 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+
+import logging
+from typing import Union
+
+from .formatter import ConsoleFormatter
+from .ansi import ANSI
+
+
+def create_record(name: str, level: int, msg: Union[str, dict, list]) -> logging.LogRecord:
+    record = logging.LogRecord(
+        name=name,
+        level=level,
+        pathname="test.py",
+        lineno=1,
+        msg=msg,
+        args=(),
+        exc_info=None
+    )
+    record.levelname = logging.getLevelName(level)
+    return record
+
+def test_error_formatting():
+    formatter = ConsoleFormatter()
+    record = create_record("test", logging.ERROR, "Error message")
+    
+    result = formatter.format(record)
+    expected = f"{ANSI.FOREGROUND_RED}{ANSI.BOLD}[ERROR] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Error message"
+    assert result == expected
+
+def test_warning_formatting():
+    formatter = ConsoleFormatter()
+    record = create_record("test", logging.WARNING, "Warning message")
+    
+    result = formatter.format(record)
+    expected = f"{ANSI.FOREGROUND_YELLOW}{ANSI.BOLD}[WARNING] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Warning message"
+    assert result == expected
+
+def test_info_formatting():
+    formatter = ConsoleFormatter()
+    record = create_record("test", logging.INFO, "Info message")
+    
+    result = formatter.format(record)
+    expected = f"{ANSI.FOREGROUND_CYAN}{ANSI.BOLD}[INFO] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Info message"
+    assert result == expected
+
+def test_debug_formatting():
+    formatter = ConsoleFormatter()
+    record = create_record("test", logging.DEBUG, "Debug message")
+    
+    result = formatter.format(record)
+    expected = f"{ANSI.FOREGROUND_MAGENTA}{ANSI.BOLD}[DEBUG] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Debug message"
+    assert result == expected
+
+def test_dict_message_formatting():
+    formatter = ConsoleFormatter()
+    dict_msg = {"key": "value", "nested": {"inner": "data"}}
+    record = create_record("test", logging.INFO, dict_msg)
+    
+    result = formatter.format(record)
+    result_lines = result.split("\n")
+    
+    prefix = f"{ANSI.FOREGROUND_CYAN}{ANSI.BOLD}[INFO] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET}"
+    assert result_lines[0].startswith(prefix)
+    assert '"key": "value"' in result
+    assert '"nested": {' in result
+    assert '"inner": "data"' in result
+
+def test_list_message_formatting():
+    formatter = ConsoleFormatter()
+    list_msg = ["item1", "item2", {"key": "value"}]
+    record = create_record("test", logging.INFO, list_msg)
+    
+    result = formatter.format(record)
+    result_lines = result.split("\n")
+    
+    prefix = f"{ANSI.FOREGROUND_CYAN}{ANSI.BOLD}[INFO] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET}"
+    assert result_lines[0].startswith(prefix)
+    assert '"item1"' in result
+    assert '"item2"' in result
+    assert '"key": "value"' in result
+
+def test_multiline_message_formatting():
+    formatter = ConsoleFormatter()
+    multiline_msg = "Line 1\nLine 2\nLine 3"
+    record = create_record("test", logging.INFO, multiline_msg)
+    
+    result = formatter.format(record)
+    expected_lines = [
+        f"{ANSI.FOREGROUND_CYAN}{ANSI.BOLD}[INFO] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Line 1",
+        f"{ANSI.FOREGROUND_CYAN}{ANSI.BOLD}[INFO] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Line 2",
+        f"{ANSI.FOREGROUND_CYAN}{ANSI.BOLD}[INFO] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Line 3"
+    ]
+    expected = "\n".join(expected_lines)
+    assert result == expected
+
+def test_unknown_level_formatting():
+    formatter = ConsoleFormatter()
+    record = create_record("test", 123, "Custom level message")
+    
+    result = formatter.format(record)
+    expected = f"{ANSI.BOLD}[Level 123] test{ANSI.FOREGROUND_RESET}{ANSI.BOLD_RESET} Custom level message"
+    assert result.endswith("Custom level message")
+    assert "[LEVEL 123]" in result
+    assert "test" in result

packages/common/src/common/storage/__init__.py

@@ -0,0 +1,10 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from .storage import Storage, ListStorage
+from .local_storage import LocalStorage
+from .list_local_storage import ListLocalStorage
+
+__all__ = ["Storage", "ListStorage", "LocalStorage", "ListLocalStorage"]

packages/common/src/common/storage/list_local_storage.py

@@ -0,0 +1,73 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+
+from __future__ import annotations
+
+from .storage import ListStorage
+from typing import TypeVar, Optional, List, Callable
+from .storage import ListStorage
+
+V = TypeVar('V')
+
+
+class ListLocalStorage(ListStorage[V]):
+
+    @property
+    def list(self) -> List[V]:
+        return self._list
+    
+    def __init__(self, items: Optional[List[V]] = []):
+        self._items = items or []
+
+    def get(self, index: int) -> Optional[V]:
+        if index < 0 or index >= len(self._items):
+            return None
+        return self._items[index]
+    
+    async def async_get(self, index: int) -> Optional[V]:
+        return await self.get(index)
+
+    def set(self, index: int, value: V) -> None:
+        self._items[index] = value
+    
+    async def async_set(self, index: int, value: V) -> None:
+        return await self.set(index, value)
+
+    def delete(self, index: int) -> None:
+        del self._items[index]
+    
+    async def async_delete(self, index: int) -> None:
+        return await self.delete(index)
+
+    def append(self, value: V) -> None:
+        return self._items.append(value)
+    
+    async def async_append(self, value: V) -> None:
+        return await self.append(value)
+
+    def pop(self) -> Optional[V]:
+        return self._items.pop()
+
+    async def async_pop(self) -> Optional[V]:
+        return await self.pop()
+
+    def items(self) -> List[V]:
+        return self._items
+
+    async def async_items(self) -> List[V]:
+        return await self.items()
+
+    def length(self) -> int:
+        return len(self._items)
+    
+    async def async_length(self) -> int:
+        return self.length()
+
+    def filter(self, predicate: Callable[[V, int], bool]) -> List[V]:
+        return [item for i, item in enumerate(self._items) if predicate(item, i)]
+
+    async def async_filter(self, predicate: Callable[[V, int], bool]) -> List[V]:
+        return await self.filter(predicate)

packages/common/src/common/storage/local_storage.py

@@ -0,0 +1,80 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+
+from __future__ import annotations
+
+from typing import Dict, List, TypeVar, Optional
+from collections import OrderedDict
+
+from pydantic import BaseModel, Field
+
+from .storage import Storage
+
+V = TypeVar('V')
+
+
+class LocalStorageOptions(BaseModel):
+    max: Optional[int] = Field(default=None, frozen=True)
+
+class LocalStorage(Storage[str, V]):
+    """A key-value storage with optional size limit and LRU behavior.
+
+    When max is set, implements LRU (Least Recently Used) behavior.
+    """
+
+    @property
+    def store(self) -> OrderedDict[str, V]:
+        return self._store
+    
+    @property
+    def options(self) -> LocalStorageOptions:
+        return self._options
+        
+    @property
+    def keys(self) -> List[str]:
+        return list(self._store.keys())
+
+    @property
+    def size(self) -> int:
+        return len(self._store)
+    
+    def __init__(
+        self, 
+        data: Optional[Dict[str, V]] = None, 
+        options: Optional[LocalStorageOptions] = None,
+    ):
+        self._store = OrderedDict(data or {})
+        self._options = options or LocalStorageOptions()
+
+    def get(self, key: str) -> Optional[V]:
+        if key not in self._store:
+            return None
+        
+        value = self._store.pop(key)
+        self._store[key] = value
+        return value
+    
+    async def async_get(self, key: str) -> Optional[V]:
+        return await self.get(key)
+
+    def set(self, key: str, value: V) -> None:
+        if key in self._store:
+            del self._store[key]
+        elif self._options.max and len(self._store) >= self._options.max:
+            self._store.popitem(last=False)
+            
+        self._store[key] = value
+    
+    async def async_set(self, key: str, value: V) -> None:
+        return await self.set(key, value)
+
+
+    def delete(self, key: str) -> None:
+        if key in self._store:
+            del self._store[key]
+
+    async def async_delete(self, key: str) -> None:
+        return await self.delete(key)