Compare commits
1 Commits
step35/667
...
fix/553-pa
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8cdae49f48 |
@@ -18,8 +18,8 @@ from urllib import request
|
||||
DEFAULT_BASE_URL = "https://forge.alexanderwhitestone.com/api/v1"
|
||||
DEFAULT_OWNER = "Timmy_Foundation"
|
||||
DEFAULT_REPO = "timmy-home"
|
||||
DEFAULT_TOKEN_FILE = Path.home() / ".config" / "gitea" / "token"
|
||||
DEFAULT_FAILOVER_STATUS = Path.home() / ".timmy" / "failover_status.json"
|
||||
DEFAULT_TOKEN_FILE = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "gitea" / "token"
|
||||
DEFAULT_FAILOVER_STATUS = Path(os.environ.get("TIMMY_HOME", Path.home() / ".timmy")) / "failover_status.json"
|
||||
DEFAULT_RESTART_STATE_DIR = Path("/var/lib/timmy/restarts")
|
||||
DEFAULT_HEARTBEAT_FILE = Path("/var/lib/timmy/heartbeats/fleet_health.last")
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ from pathlib import Path
|
||||
|
||||
|
||||
def get_token():
|
||||
f = Path.home() / ".config" / "gitea" / "token"
|
||||
f = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "gitea" / "token"
|
||||
if f.exists():
|
||||
return f.read_text().strip()
|
||||
return os.environ.get("GITEA_TOKEN", "")
|
||||
|
||||
@@ -15,7 +15,7 @@ from typing import Any, Dict, List
|
||||
|
||||
# Configuration
|
||||
GITEA_BASE = "https://forge.alexanderwhitestone.com/api/v1"
|
||||
TOKEN_PATH = os.path.expanduser("~/.config/gitea/token")
|
||||
TOKEN_PATH = os.path.join(os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")), "gitea", "token")
|
||||
ORG = "Timmy_Foundation"
|
||||
REPO = "timmy-home"
|
||||
|
||||
|
||||
@@ -13,7 +13,7 @@ from urllib.request import Request, urlopen
|
||||
|
||||
API_BASE = "https://forge.alexanderwhitestone.com/api/v1"
|
||||
ORG = "Timmy_Foundation"
|
||||
DEFAULT_TOKEN_PATH = os.path.expanduser("~/.config/gitea/token")
|
||||
DEFAULT_TOKEN_PATH = os.path.join(os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")), "gitea", "token")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
||||
@@ -143,176 +143,66 @@ def generate_test(gap):
|
||||
lines = []
|
||||
lines.append(f" # AUTO-GENERATED -- review before merging")
|
||||
lines.append(f" # Source: {func.module_path}:{func.lineno}")
|
||||
lines.append(f" # Function: {func.qualified_name}")
|
||||
lines.append("")
|
||||
mod_imp = func.module_path.replace("/", ".").replace("-", "_").replace(".py", "")
|
||||
|
||||
# Build arguments
|
||||
call_args = []
|
||||
for a in func.args:
|
||||
if a in ("self", "cls"):
|
||||
continue
|
||||
if "path" in a or "file" in a or "dir" in a:
|
||||
call_args.append(f"{a}='/tmp/test'")
|
||||
elif "name" in a or "id" in a or "key" in a:
|
||||
call_args.append(f"{a}='test'")
|
||||
elif "message" in a or "text" in a:
|
||||
call_args.append(f"{a}='test msg'")
|
||||
elif "count" in a or "num" in a or "size" in a or "width" in a or "height" in a:
|
||||
call_args.append(f"{a}=1")
|
||||
elif "flag" in a or "enabled" in a or "verbose" in a:
|
||||
call_args.append(f"{a}=False")
|
||||
else:
|
||||
call_args.append(f"{a}=MagicMock()")
|
||||
if a in ("self", "cls"): continue
|
||||
if "path" in a or "file" in a or "dir" in a: call_args.append(f"{a}='/tmp/test'")
|
||||
elif "name" in a: call_args.append(f"{a}='test'")
|
||||
elif "id" in a or "key" in a: call_args.append(f"{a}='test_id'")
|
||||
elif "message" in a or "text" in a: call_args.append(f"{a}='test msg'")
|
||||
elif "count" in a or "num" in a or "size" in a: call_args.append(f"{a}=1")
|
||||
elif "flag" in a or "enabled" in a or "verbose" in a: call_args.append(f"{a}=False")
|
||||
else: call_args.append(f"{a}=None")
|
||||
args_str = ", ".join(call_args)
|
||||
|
||||
# Test function header
|
||||
if func.is_async:
|
||||
lines.append(" @pytest.mark.asyncio")
|
||||
lines.append(f" async def {func.test_name}(self):")
|
||||
else:
|
||||
lines.append(f" def {func.test_name}(self):")
|
||||
|
||||
lines.append(f" def {func.test_name}(self):")
|
||||
lines.append(f' """Test {func.qualified_name} -- auto-generated."""')
|
||||
|
||||
if func.class_name:
|
||||
lines.append(" try:")
|
||||
lines.append(f" try:")
|
||||
lines.append(f" from {mod_imp} import {func.class_name}")
|
||||
if func.is_private:
|
||||
lines.append(" pytest.skip('Private method')")
|
||||
lines.append(f" pytest.skip('Private method')")
|
||||
elif func.is_property:
|
||||
lines.append(f" obj = {func.class_name}()")
|
||||
lines.append(f" _ = obj.{func.name}")
|
||||
else:
|
||||
if func.raises:
|
||||
lines.append(f" with pytest.raises(({', '.join(func.raises)})):")
|
||||
if func.is_async:
|
||||
lines.append(f" await {func.class_name}().{func.name}({args_str})")
|
||||
else:
|
||||
lines.append(f" {func.class_name}().{func.name}({args_str})")
|
||||
lines.append(f" {func.class_name}().{func.name}({args_str})")
|
||||
else:
|
||||
lines.append(f" obj = {func.class_name}()")
|
||||
if func.is_async:
|
||||
lines.append(f" _ = await obj.{func.name}({args_str})")
|
||||
else:
|
||||
lines.append(f" _ = obj.{func.name}({args_str})")
|
||||
lines.append(" except ImportError:")
|
||||
lines.append(" pytest.skip('Module not importable')")
|
||||
lines.append(f" result = obj.{func.name}({args_str})")
|
||||
if func.has_return:
|
||||
lines.append(f" assert result is not None or result is None # Placeholder")
|
||||
lines.append(f" except ImportError:")
|
||||
lines.append(f" pytest.skip('Module not importable')")
|
||||
else:
|
||||
lines.append(" try:")
|
||||
lines.append(f" try:")
|
||||
lines.append(f" from {mod_imp} import {func.name}")
|
||||
if func.is_private:
|
||||
lines.append(" pytest.skip('Private function')")
|
||||
lines.append(f" pytest.skip('Private function')")
|
||||
else:
|
||||
if func.raises:
|
||||
lines.append(f" with pytest.raises(({', '.join(func.raises)})):")
|
||||
if func.is_async:
|
||||
lines.append(f" await {func.name}({args_str})")
|
||||
else:
|
||||
lines.append(f" {func.name}({args_str})")
|
||||
lines.append(f" {func.name}({args_str})")
|
||||
else:
|
||||
if func.is_async:
|
||||
lines.append(f" _ = await {func.name}({args_str})")
|
||||
else:
|
||||
lines.append(f" _ = {func.name}({args_str})")
|
||||
lines.append(" except ImportError:")
|
||||
lines.append(" pytest.skip('Module not importable')")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def generate_edge_cases(gap):
|
||||
"""Generate edge case test for a function."""
|
||||
func = gap.func
|
||||
lines = []
|
||||
lines.append(f" # AUTO-GENERATED -- edge cases -- review before merging")
|
||||
lines.append(f" # Source: {func.module_path}:{func.lineno}")
|
||||
lines.append("")
|
||||
mod_imp = func.module_path.replace("/", ".").replace("-", "_").replace(".py", "")
|
||||
test_name = f"{func.test_name}_edge_cases"
|
||||
|
||||
if func.is_async:
|
||||
lines.append(" @pytest.mark.asyncio")
|
||||
lines.append(f" async def {test_name}(self):")
|
||||
else:
|
||||
lines.append(f" def {test_name}(self):")
|
||||
|
||||
lines.append(f' """Edge cases for {func.qualified_name}."""')
|
||||
|
||||
# Edge argument values
|
||||
call_args = []
|
||||
for a in func.args:
|
||||
if a in ("self", "cls"):
|
||||
continue
|
||||
if "path" in a or "file" in a or "dir" in a:
|
||||
call_args.append(f"{a}=''")
|
||||
elif "name" in a or "id" in a or "key" in a:
|
||||
call_args.append(f"{a}=''")
|
||||
elif "message" in a or "text" in a:
|
||||
call_args.append(f"{a}=''")
|
||||
elif "count" in a or "num" in a or "size" in a or "width" in a or "height" in a:
|
||||
call_args.append(f"{a}=0")
|
||||
elif "flag" in a or "enabled" in a or "verbose" in a:
|
||||
call_args.append(f"{a}=False")
|
||||
else:
|
||||
call_args.append(f"{a}=MagicMock()")
|
||||
args_str = ", ".join(call_args)
|
||||
|
||||
if func.class_name:
|
||||
lines.append(" try:")
|
||||
lines.append(f" from {mod_imp} import {func.class_name}")
|
||||
lines.append(f" obj = {func.class_name}()")
|
||||
if func.is_async:
|
||||
lines.append(f" _ = await obj.{func.name}({args_str})")
|
||||
else:
|
||||
lines.append(f" _ = obj.{func.name}({args_str})")
|
||||
lines.append(" except ImportError:")
|
||||
lines.append(" pytest.skip('Module not importable')")
|
||||
else:
|
||||
lines.append(" try:")
|
||||
lines.append(f" from {mod_imp} import {func.name}")
|
||||
if func.is_async:
|
||||
lines.append(f" _ = await {func.name}({args_str})")
|
||||
else:
|
||||
lines.append(f" _ = {func.name}({args_str})")
|
||||
lines.append(" except ImportError:")
|
||||
lines.append(" pytest.skip('Module not importable')")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def generate_test_suite(gaps, max_tests=50):
|
||||
by_module = {}
|
||||
for gap in gaps[:max_tests]:
|
||||
by_module.setdefault(gap.func.module_path, []).append(gap)
|
||||
|
||||
lines = []
|
||||
lines.append('"""Auto-generated test suite -- Codebase Genome (#667).')
|
||||
lines.append("")
|
||||
lines.append("Generated by scripts/codebase_test_generator.py")
|
||||
lines.append("Coverage gaps identified from AST analysis.")
|
||||
lines.append("")
|
||||
lines.append("These tests are starting points. Review before merging.")
|
||||
lines.append('"""')
|
||||
lines.append("")
|
||||
lines.append("import pytest")
|
||||
lines.append("from unittest.mock import MagicMock, patch")
|
||||
lines.append("")
|
||||
lines.append("")
|
||||
lines.append("# AUTO-GENERATED -- DO NOT EDIT WITHOUT REVIEW")
|
||||
|
||||
for module, mgaps in sorted(by_module.items()):
|
||||
safe = module.replace("/", "_").replace(".py", "").replace("-", "_")
|
||||
cls_name = "".join(w.title() for w in safe.split("_"))
|
||||
lines.append("")
|
||||
lines.append(f"class Test{cls_name}Generated:")
|
||||
lines.append(f' """Auto-generated tests for {module}."""')
|
||||
for gap in mgaps:
|
||||
lines.append("")
|
||||
lines.append(generate_test(gap))
|
||||
lines.append(generate_edge_cases(gap))
|
||||
lines.append("")
|
||||
lines.append(f" result = {func.name}({args_str})")
|
||||
if func.has_return:
|
||||
lines.append(f" assert result is not None or result is None # Placeholder")
|
||||
lines.append(f" except ImportError:")
|
||||
lines.append(f" pytest.skip('Module not importable')")
|
||||
|
||||
return chr(10).join(lines)
|
||||
|
||||
|
||||
def generate_test_suite(gaps, max_tests=50):
|
||||
by_module = {}
|
||||
for gap in gaps[:max_tests]:
|
||||
by_module.setdefault(gap.func.module_path, []).append(gap)
|
||||
@@ -386,7 +276,7 @@ def main():
|
||||
return
|
||||
|
||||
if gaps:
|
||||
content = generate_test_suite(gaps, max_tests=args.max_tests)
|
||||
content = generate_test_suite(gaps, max_tests=args.max-tests if hasattr(args, 'max-tests') else args.max_tests)
|
||||
out = os.path.join(source_dir, args.output)
|
||||
os.makedirs(os.path.dirname(out), exist_ok=True)
|
||||
with open(out, "w") as f:
|
||||
|
||||
@@ -27,7 +27,7 @@ from pathlib import Path
|
||||
import re
|
||||
|
||||
GITEA_URL = "https://forge.alexanderwhitestone.com"
|
||||
GITEA_TOKEN_PATH = Path.home() / ".config" / "gitea" / "token"
|
||||
GITEA_TOKEN_PATH = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "gitea" / "token"
|
||||
ORG = "Timmy_Foundation"
|
||||
|
||||
REPOS = [
|
||||
|
||||
@@ -12,12 +12,14 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
STATUS_FILE = Path.home() / ".timmy" / "failover_status.json"
|
||||
SPEC_FILE = Path.home() / ".timmy" / "fleet_dispatch.json"
|
||||
OUTPUT_FILE = Path.home() / ".timmy" / "dispatch_plan.json"
|
||||
TIMMY_HOME = Path(os.environ.get("TIMMY_HOME", Path.home() / ".timmy"))
|
||||
STATUS_FILE = TIMMY_HOME / "failover_status.json"
|
||||
SPEC_FILE = TIMMY_HOME / "fleet_dispatch.json"
|
||||
OUTPUT_FILE = TIMMY_HOME / "dispatch_plan.json"
|
||||
|
||||
|
||||
def load_json(path: Path, default: Any):
|
||||
|
||||
@@ -13,7 +13,7 @@ FLEET = {
|
||||
"bezalel": "167.99.126.228"
|
||||
}
|
||||
|
||||
STATUS_FILE = Path.home() / ".timmy" / "failover_status.json"
|
||||
STATUS_FILE = Path(os.environ.get("TIMMY_HOME", Path.home() / ".timmy")) / "failover_status.json"
|
||||
|
||||
def check_health(host):
|
||||
try:
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user