mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-13 08:15:59 +00:00
Refactoring in tests: prepare for testing with multiple languages
This commit is contained in:
1 parent
072b5caefb
commit
433d7f4d59
24 files changed
+103
-159
No files matched your search
@@ -5,8 +5,8 @@ it assumes that you already have a serena_config.yml file setup.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pprint import pprint
|
||||
from pathlib import Path
|
||||
from pprint import pprint
|
||||
|
||||
from serena.agent import *
|
||||
|
||||
@@ -16,11 +16,11 @@ if __name__ == "__main__":
|
||||
agent = SerenaAgent(project_file_path=str(project_root / ".serena" / "project.yml"))
|
||||
overview_tool = agent.get_tool(GetSymbolsOverviewTool)
|
||||
find_symbol_tool = agent.get_tool(FindSymbolTool)
|
||||
|
||||
|
||||
print("Getting an overview of the util package\n")
|
||||
pprint(json.loads(overview_tool.apply("src/serena/util")))
|
||||
|
||||
|
||||
print("\n\n")
|
||||
|
||||
|
||||
print("Finding the symbol 'SerenaAgent'\n")
|
||||
pprint(json.loads(find_symbol_tool.apply("SerenaAgent")))
|
||||
pprint(json.loads(find_symbol_tool.apply("SerenaAgent")))
|
||||
File renamed without changes.
+59
-12
@@ -14,25 +14,72 @@ def resources_dir() -> Path:
|
||||
return current_dir / "resources"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def repo_path() -> Path:
|
||||
return Path(__file__).parent / "resources" / "test_repo"
|
||||
class LanguageParamRequest:
|
||||
param: Language
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def language_server(repo_path: Path):
|
||||
"""Create a SyncLanguageServer instance configured to use the test repository."""
|
||||
config = MultilspyConfig(code_language=Language.PYTHON)
|
||||
def get_repo_path(language: Language) -> Path:
|
||||
return Path(__file__).parent / "resources" / "repos" / language / "test_repo"
|
||||
|
||||
|
||||
def create_default_ls(language: Language) -> SyncLanguageServer:
|
||||
config = MultilspyConfig(code_language=language)
|
||||
repo_path = str(get_repo_path(language))
|
||||
logger = MultilspyLogger()
|
||||
return SyncLanguageServer.create(config, logger, repo_path)
|
||||
|
||||
# Create a language server instance
|
||||
server = SyncLanguageServer.create(config, logger, str(repo_path))
|
||||
|
||||
# Start the server
|
||||
@pytest.fixture(scope="session")
|
||||
def repo_path(request: LanguageParamRequest) -> Path:
|
||||
"""Get the repository path for a specific language.
|
||||
|
||||
This fixture requires a language parameter via pytest.mark.parametrize:
|
||||
|
||||
Example:
|
||||
```
|
||||
@pytest.mark.parametrize("repo_path", [Language.PYTHON], indirect=True)
|
||||
def test_python_repo(repo_path):
|
||||
assert (repo_path / "src").exists()
|
||||
```
|
||||
|
||||
"""
|
||||
if not hasattr(request, "param"):
|
||||
raise ValueError("Language parameter must be provided via pytest.mark.parametrize")
|
||||
|
||||
language = request.param
|
||||
return get_repo_path(language)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def language_server(request: LanguageParamRequest):
|
||||
"""Create a SyncLanguageServer instance configured for the specified language.
|
||||
|
||||
This fixture requires a language parameter via pytest.mark.parametrize:
|
||||
|
||||
Example:
|
||||
```
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_python_server(language_server):
|
||||
# Use the Python language server
|
||||
pass
|
||||
```
|
||||
|
||||
You can also test multiple languages in a single test:
|
||||
```
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON, Language.TYPESCRIPT], indirect=True)
|
||||
def test_multiple_languages(language_server):
|
||||
# This test will run once for each language
|
||||
pass
|
||||
```
|
||||
|
||||
"""
|
||||
if not hasattr(request, "param"):
|
||||
raise ValueError("Language parameter must be provided via pytest.mark.parametrize")
|
||||
|
||||
language = request.param
|
||||
server = create_default_ls(language)
|
||||
server.start()
|
||||
|
||||
try:
|
||||
yield server
|
||||
finally:
|
||||
# Ensure server is shut down
|
||||
server.stop()
|
||||
@@ -7,13 +7,17 @@ like request_references using the test repository.
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from multilspy.language_server import SyncLanguageServer
|
||||
from multilspy.multilspy_config import Language
|
||||
from serena.text_utils import LineType
|
||||
|
||||
|
||||
class TestLanguageServerBasics:
|
||||
"""Test basic functionality of the language server."""
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_references_user_class(self, language_server: SyncLanguageServer):
|
||||
"""Test request_references on the User class."""
|
||||
# Get references to the User class in models.py
|
||||
@@ -27,6 +31,7 @@ class TestLanguageServerBasics:
|
||||
# At least two references should be found (one for the class definition itself)
|
||||
assert len(references) > 1
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_references_item_class(self, language_server: SyncLanguageServer):
|
||||
"""Test request_references on the Item class."""
|
||||
# Get references to the Item class in models.py
|
||||
@@ -41,6 +46,7 @@ class TestLanguageServerBasics:
|
||||
services_references = [ref for ref in references if "services.py" in ref["uri"]]
|
||||
assert len(services_references) > 0
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_references_function_parameter(self, language_server: SyncLanguageServer):
|
||||
"""Test request_references on a function parameter."""
|
||||
# Get references to the id parameter in get_user method
|
||||
@@ -51,6 +57,7 @@ class TestLanguageServerBasics:
|
||||
# id parameter should be referenced within the method
|
||||
assert len(references) > 0
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_references_create_user_method(self, language_server: SyncLanguageServer):
|
||||
# Get references to the create_user method in UserService
|
||||
file_path = os.path.join("test_repo", "services.py")
|
||||
@@ -60,6 +67,7 @@ class TestLanguageServerBasics:
|
||||
# Verify that we get valid references
|
||||
assert len(references) > 1
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_retrieve_content_around_line(self, language_server: SyncLanguageServer):
|
||||
"""Test retrieve_content_around_line functionality with various scenarios."""
|
||||
file_path = os.path.join("test_repo", "models.py")
|
||||
@@ -166,6 +174,7 @@ class TestLanguageServerBasics:
|
||||
else:
|
||||
assert line.match_type == LineType.AFTER_MATCH
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_search_files_for_pattern(self, language_server: SyncLanguageServer):
|
||||
"""Test search_files_for_pattern with various patterns and glob filters."""
|
||||
# Test 1: Search for class definitions across all files
|
||||
+6
-1
@@ -6,10 +6,11 @@ import pytest
|
||||
from multilspy.language_server import SyncLanguageServer
|
||||
from multilspy.multilspy_config import Language, MultilspyConfig
|
||||
from multilspy.multilspy_logger import MultilspyLogger
|
||||
from test.conftest import get_repo_path
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ls_with_ignored_dirs(repo_path: Path) -> Generator[SyncLanguageServer, None, None]:
|
||||
def ls_with_ignored_dirs() -> Generator[SyncLanguageServer, None, None]:
|
||||
"""Fixture to set up an LS for the python test repo with the 'scripts' directory ignored."""
|
||||
config = MultilspyConfig(
|
||||
code_language=Language.PYTHON,
|
||||
@@ -17,6 +18,7 @@ def ls_with_ignored_dirs(repo_path: Path) -> Generator[SyncLanguageServer, None,
|
||||
ignored_paths=["scripts", "custom_test"], # Configure the relative path to be ignored
|
||||
)
|
||||
logger = MultilspyLogger()
|
||||
repo_path = get_repo_path(Language.PYTHON)
|
||||
ls = SyncLanguageServer.create(config, logger, str(repo_path))
|
||||
ls.start()
|
||||
try:
|
||||
@@ -25,6 +27,7 @@ def ls_with_ignored_dirs(repo_path: Path) -> Generator[SyncLanguageServer, None,
|
||||
ls.stop()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ls_with_ignored_dirs", [Language.PYTHON], indirect=True)
|
||||
def test_symbol_tree_ignores_dir(ls_with_ignored_dirs: SyncLanguageServer):
|
||||
"""Tests that request_full_symbol_tree ignores the configured directory."""
|
||||
root = ls_with_ignored_dirs.request_full_symbol_tree()[0]
|
||||
@@ -33,6 +36,7 @@ def test_symbol_tree_ignores_dir(ls_with_ignored_dirs: SyncLanguageServer):
|
||||
assert children_names == {"test_repo", "examples"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ls_with_ignored_dirs", [Language.PYTHON], indirect=True)
|
||||
def test_find_references_ignores_dir(ls_with_ignored_dirs: SyncLanguageServer):
|
||||
"""Tests that find_references ignores the configured directory."""
|
||||
# Location of Item, which is referenced in scripts
|
||||
@@ -46,6 +50,7 @@ def test_find_references_ignores_dir(ls_with_ignored_dirs: SyncLanguageServer):
|
||||
assert not any("scripts" in ref["relativePath"] for ref in references)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("repo_path", [Language.PYTHON], indirect=True)
|
||||
def test_refs_and_symbols_with_glob_patterns(repo_path: Path) -> None:
|
||||
"""Tests that refs and symbols with glob patterns are ignored."""
|
||||
config = MultilspyConfig(
|
||||
+24
@@ -8,13 +8,17 @@ These tests focus on the following methods:
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from multilspy.language_server import SyncLanguageServer
|
||||
from multilspy.multilspy_config import Language
|
||||
from multilspy.multilspy_types import SymbolKind
|
||||
|
||||
|
||||
class TestLanguageServerSymbols:
|
||||
"""Test the language server's symbol-related functionality."""
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_containing_symbol_function(self, language_server: SyncLanguageServer):
|
||||
"""Test request_containing_symbol for a function."""
|
||||
# Test for a position inside the create_user method
|
||||
@@ -29,6 +33,7 @@ class TestLanguageServerSymbols:
|
||||
if "body" in containing_symbol:
|
||||
assert containing_symbol["body"].strip().startswith("def create_user(self")
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_references_to_variables(self, language_server: SyncLanguageServer):
|
||||
"""Test request_referencing_symbols for a variable."""
|
||||
file_path = os.path.join("test_repo", "variables.py")
|
||||
@@ -43,6 +48,7 @@ class TestLanguageServerSymbols:
|
||||
assert "dataclass_instance" in ref_names
|
||||
assert "second_dataclass" in ref_names
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_containing_symbol_class(self, language_server: SyncLanguageServer):
|
||||
"""Test request_containing_symbol for a class."""
|
||||
# Test for a position inside the UserService class but outside any method
|
||||
@@ -55,6 +61,7 @@ class TestLanguageServerSymbols:
|
||||
assert containing_symbol["name"] == "UserService"
|
||||
assert containing_symbol["kind"] == SymbolKind.Class
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_containing_symbol_nested(self, language_server: SyncLanguageServer):
|
||||
"""Test request_containing_symbol with nested scopes."""
|
||||
# Test for a position inside a method which is inside a class
|
||||
@@ -80,6 +87,7 @@ class TestLanguageServerSymbols:
|
||||
assert parent_symbol["name"] == "UserService"
|
||||
assert parent_symbol["kind"] == SymbolKind.Class
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_containing_symbol_none(self, language_server: SyncLanguageServer):
|
||||
"""Test request_containing_symbol for a position with no containing symbol."""
|
||||
# Test for a position outside any function/class (e.g., in imports)
|
||||
@@ -90,6 +98,7 @@ class TestLanguageServerSymbols:
|
||||
# Should return None or an empty dictionary
|
||||
assert containing_symbol is None or containing_symbol == {}
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_referencing_symbols_function(self, language_server: SyncLanguageServer):
|
||||
"""Test request_referencing_symbols for a function."""
|
||||
# Test referencing symbols for create_user function
|
||||
@@ -108,6 +117,7 @@ class TestLanguageServerSymbols:
|
||||
assert "start" in symbol["location"]["range"]
|
||||
assert "end" in symbol["location"]["range"]
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_referencing_symbols_class(self, language_server: SyncLanguageServer):
|
||||
"""Test request_referencing_symbols for a class."""
|
||||
# Test referencing symbols for User class
|
||||
@@ -127,6 +137,7 @@ class TestLanguageServerSymbols:
|
||||
|
||||
assert len(services_references) > 0
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_referencing_symbols_parameter(self, language_server: SyncLanguageServer):
|
||||
"""Test request_referencing_symbols for a function parameter."""
|
||||
# Test referencing symbols for id parameter in get_user
|
||||
@@ -146,6 +157,7 @@ class TestLanguageServerSymbols:
|
||||
|
||||
assert len(method_refs) > 0
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_referencing_symbols_none(self, language_server: SyncLanguageServer):
|
||||
"""Test request_referencing_symbols for a position with no symbol."""
|
||||
# For positions with no symbol, the method might throw an error or return None/empty list
|
||||
@@ -163,6 +175,7 @@ class TestLanguageServerSymbols:
|
||||
pass
|
||||
|
||||
# Tests for request_defining_symbol
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_defining_symbol_variable(self, language_server: SyncLanguageServer):
|
||||
"""Test request_defining_symbol for a variable usage."""
|
||||
# Test finding the definition of a symbol in the create_user method
|
||||
@@ -181,6 +194,7 @@ class TestLanguageServerSymbols:
|
||||
if "location" in defining_symbol and "uri" in defining_symbol["location"]:
|
||||
assert "services.py" in defining_symbol["location"]["uri"]
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_defining_symbol_imported_class(self, language_server: SyncLanguageServer):
|
||||
"""Test request_defining_symbol for an imported class."""
|
||||
# Test finding the definition of the 'User' class used in the UserService.create_user method
|
||||
@@ -192,6 +206,7 @@ class TestLanguageServerSymbols:
|
||||
assert defining_symbol is not None
|
||||
assert defining_symbol.get("name") == "User"
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_defining_symbol_method_call(self, language_server: SyncLanguageServer):
|
||||
"""Test request_defining_symbol for a method call."""
|
||||
# Create an example file path for a file that calls UserService.create_user
|
||||
@@ -216,6 +231,7 @@ class TestLanguageServerSymbols:
|
||||
|
||||
warnings.warn("Could not verify method call definition - file structure may differ from expected")
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_defining_symbol_none(self, language_server: SyncLanguageServer):
|
||||
"""Test request_defining_symbol for a position with no symbol."""
|
||||
# Test for a position with no symbol (e.g., whitespace or comment)
|
||||
@@ -226,6 +242,7 @@ class TestLanguageServerSymbols:
|
||||
# Should return None for positions with no symbol
|
||||
assert defining_symbol is None
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_containing_symbol_variable(self, language_server: SyncLanguageServer):
|
||||
"""Test request_containing_symbol where the symbol is a variable."""
|
||||
# Test for a position inside a variable definition
|
||||
@@ -238,6 +255,7 @@ class TestLanguageServerSymbols:
|
||||
assert containing_symbol["name"] == "user_var_str"
|
||||
assert containing_symbol["kind"] == SymbolKind.Variable
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_defining_symbol_nested_function(self, language_server: SyncLanguageServer):
|
||||
"""Test request_defining_symbol for a nested function or closure."""
|
||||
# Use the existing nested.py file which contains nested classes and methods
|
||||
@@ -288,6 +306,7 @@ class TestLanguageServerSymbols:
|
||||
assert defining_symbol.get("name") == "func_within_func"
|
||||
assert defining_symbol.get("kind") == SymbolKind.Function.value
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_symbol_methods_integration(self, language_server: SyncLanguageServer):
|
||||
"""Test the integration between different symbol-related methods."""
|
||||
# This test demonstrates using the various symbol methods together
|
||||
@@ -336,6 +355,7 @@ class TestLanguageServerSymbols:
|
||||
|
||||
warnings.warn("Could not verify container hierarchy - implementation detail")
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_symbol_tree_structure(self, language_server: SyncLanguageServer):
|
||||
"""Test that the symbol tree structure is correctly built."""
|
||||
# Get all symbols in the test file
|
||||
@@ -363,6 +383,7 @@ class TestLanguageServerSymbols:
|
||||
_, user_management_roots = language_server.request_document_symbols(os.path.join("examples", "user_management.py"))
|
||||
assert user_management_roots == user_management_node["children"]
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_symbol_tree_structure_subdir(self, language_server: SyncLanguageServer):
|
||||
"""Test that the symbol tree structure is correctly built."""
|
||||
# Get all symbols in the test file
|
||||
@@ -383,6 +404,7 @@ class TestLanguageServerSymbols:
|
||||
_, user_management_roots = language_server.request_document_symbols(os.path.join("examples", "user_management.py"))
|
||||
assert user_management_roots == user_management_node["children"]
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_dir_overview(self, language_server: SyncLanguageServer):
|
||||
"""Test that request_dir_overview returns correct symbol information for files in a directory."""
|
||||
# Get overview of the examples directory
|
||||
@@ -407,6 +429,7 @@ class TestLanguageServerSymbols:
|
||||
for symbol in expected_symbols:
|
||||
assert symbol in services_symbols
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_request_document_overview(self, language_server: SyncLanguageServer):
|
||||
"""Test that request_document_overview returns correct symbol information for a file."""
|
||||
# Get overview of the user_management.py file
|
||||
@@ -416,6 +439,7 @@ class TestLanguageServerSymbols:
|
||||
symbol_names = {s_info[0] for s_info in overview}
|
||||
assert {"UserStats", "UserManager", "process_user_data", "main"}.issubset(symbol_names)
|
||||
|
||||
@pytest.mark.parametrize("language_server", [Language.PYTHON], indirect=True)
|
||||
def test_containing_symbol_of_var_is_file(self, language_server: SyncLanguageServer):
|
||||
"""Test that the containing symbol of a variable is the file itself."""
|
||||
# Get the containing symbol of a variable in a file
|
||||
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
Whitespace-only changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
@@ -1,141 +0,0 @@
|
||||
"""
|
||||
Example demonstrating user management with the test_repo module.
|
||||
|
||||
This example showcases:
|
||||
- Creating and managing users
|
||||
- Using various object types and relationships
|
||||
- Type annotations and complex Python patterns
|
||||
"""
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from test_repo.models import User, create_user_object
|
||||
from test_repo.services import UserService
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class UserStats:
|
||||
"""Statistics about user activity."""
|
||||
|
||||
user_id: str
|
||||
login_count: int = 0
|
||||
last_active_days: int = 0
|
||||
engagement_score: float = 0.0
|
||||
|
||||
def is_active(self) -> bool:
|
||||
"""Check if the user is considered active."""
|
||||
return self.last_active_days < 30
|
||||
|
||||
|
||||
class UserManager:
|
||||
"""Example class demonstrating complex user management."""
|
||||
|
||||
def __init__(self, service: UserService):
|
||||
self.service = service
|
||||
self.active_users: dict[str, User] = {}
|
||||
self.user_stats: dict[str, UserStats] = {}
|
||||
|
||||
def register_user(self, name: str, email: str, roles: list[str] | None = None) -> User:
|
||||
"""Register a new user."""
|
||||
logger.info(f"Registering new user: {name} ({email})")
|
||||
user = self.service.create_user(name=name, email=email, roles=roles)
|
||||
self.active_users[user.id] = user
|
||||
self.user_stats[user.id] = UserStats(user_id=user.id)
|
||||
return user
|
||||
|
||||
def get_user(self, user_id: str) -> User | None:
|
||||
"""Get a user by ID."""
|
||||
if user_id in self.active_users:
|
||||
return self.active_users[user_id]
|
||||
|
||||
# Try to fetch from service
|
||||
user = self.service.get_user(user_id)
|
||||
if user:
|
||||
self.active_users[user.id] = user
|
||||
return user
|
||||
|
||||
def update_user_stats(self, user_id: str, login_count: int, days_since_active: int) -> None:
|
||||
"""Update statistics for a user."""
|
||||
if user_id not in self.user_stats:
|
||||
self.user_stats[user_id] = UserStats(user_id=user_id)
|
||||
|
||||
stats = self.user_stats[user_id]
|
||||
stats.login_count = login_count
|
||||
stats.last_active_days = days_since_active
|
||||
|
||||
# Calculate engagement score based on activity
|
||||
engagement = (100 - min(days_since_active, 100)) * 0.8
|
||||
engagement += min(login_count, 20) * 0.2
|
||||
stats.engagement_score = engagement
|
||||
|
||||
def get_active_users(self) -> list[User]:
|
||||
"""Get all active users."""
|
||||
active_user_ids = [user_id for user_id, stats in self.user_stats.items() if stats.is_active()]
|
||||
return [self.active_users[user_id] for user_id in active_user_ids if user_id in self.active_users]
|
||||
|
||||
def get_user_by_email(self, email: str) -> User | None:
|
||||
"""Find a user by their email address."""
|
||||
for user in self.active_users.values():
|
||||
if user.email == email:
|
||||
return user
|
||||
return None
|
||||
|
||||
|
||||
# Example function demonstrating type annotations
|
||||
def process_user_data(users: list[User], include_inactive: bool = False, transform_func: callable | None = None) -> dict[str, Any]:
|
||||
"""Process user data with optional transformations."""
|
||||
result: dict[str, Any] = {"users": [], "total": 0, "admin_count": 0}
|
||||
|
||||
for user in users:
|
||||
if transform_func:
|
||||
user_data = transform_func(user.to_dict())
|
||||
else:
|
||||
user_data = user.to_dict()
|
||||
|
||||
result["users"].append(user_data)
|
||||
result["total"] += 1
|
||||
|
||||
if "admin" in user.roles:
|
||||
result["admin_count"] += 1
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def main():
|
||||
"""Main function demonstrating the usage of UserManager."""
|
||||
# Initialize service and manager
|
||||
service = UserService()
|
||||
manager = UserManager(service)
|
||||
|
||||
# Register some users
|
||||
admin = manager.register_user("Admin User", "admin@example.com", ["admin"])
|
||||
user1 = manager.register_user("Regular User", "user@example.com", ["user"])
|
||||
user2 = manager.register_user("Another User", "another@example.com", ["user"])
|
||||
|
||||
# Update some stats
|
||||
manager.update_user_stats(admin.id, 100, 5)
|
||||
manager.update_user_stats(user1.id, 50, 10)
|
||||
manager.update_user_stats(user2.id, 10, 45) # Inactive user
|
||||
|
||||
# Get active users
|
||||
active_users = manager.get_active_users()
|
||||
logger.info(f"Active users: {len(active_users)}")
|
||||
|
||||
# Process user data
|
||||
user_data = process_user_data(active_users, transform_func=lambda u: {**u, "full_name": u.get("name", "")})
|
||||
|
||||
logger.info(f"Processed {user_data['total']} users, {user_data['admin_count']} admins")
|
||||
|
||||
# Example of calling create_user directly
|
||||
external_user = create_user_object(id="ext123", name="External User", email="external@example.org", roles=["external"])
|
||||
logger.info(f"Created external user: {external_user.name}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user