Merge pull request #26 from anurag03/ansinha/cleanup

anurag03 · web-flow · commit af2ee790dd35 · 2025-09-23T22:13:42.000+05:30
Adding pre-commit hook and standardizing repo code.
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -0,0 +1,13 @@
+repos:
+  - repo: https://github.com/psf/black
+    rev: 24.3.0
+    hooks:
+      - id: black
+        language_version: python3
+  - repo: https://github.com/pycqa/flake8
+    rev: 7.0.0
+    hooks:
+      - id: flake8
+        language_version: python3
+        additional_dependencies: ["flake8"]
+        args: ["--ignore=E501"]
diff --git a/setup.py b/setup.py
@@ -5,7 +5,7 @@
 
 setup(
     name="testcato",
-    version="1.2.2",
+    version="1.2.3",
     packages=find_packages(),
     description="A package for categorizing test results.",
     long_description=long_description,
diff --git a/testcato/__init__.py b/testcato/__init__.py
@@ -1,18 +1,19 @@
-
 # Package metadata for wheel
 __title__ = "testcato"
-__version__ = "1.2.2"
+__version__ = "1.2.3"
 
 # Automatically create config file if not present
 import os
 import shutil
 
+
 def _ensure_config():
-	config_name = "testcato_config.yaml"
-	config_src = os.path.join(os.path.dirname(__file__), "..", config_name)
-	config_dst = os.path.join(os.getcwd(), config_name)
-	if not os.path.exists(config_dst):
-		if os.path.exists(config_src):
-			shutil.copyfile(config_src, config_dst)
+    config_name = "testcato_config.yaml"
+    config_src = os.path.join(os.path.dirname(__file__), "..", config_name)
+    config_dst = os.path.join(os.getcwd(), config_name)
+    if not os.path.exists(config_dst):
+        if os.path.exists(config_src):
+            shutil.copyfile(config_src, config_dst)
+
 
 _ensure_config()
diff --git a/testcato/ai_agent.py b/testcato/ai_agent.py
@@ -1,120 +1,142 @@
 import os
 import glob
-## import xml.etree.ElementTree as ET  # No longer needed, switched to JSONL
+
+# import xml.etree.ElementTree as ET  # No longer needed, switched to JSONL
 import datetime
 import requests
 import yaml
 
+
 def get_latest_jsonl(result_dir):
-    files = glob.glob(os.path.join(result_dir, 'test_run_*.jsonl'))
+    files = glob.glob(os.path.join(result_dir, "test_run_*.jsonl"))
     if not files:
         return None
     return max(files, key=os.path.getctime)
 
+
 def load_agent_config(config_path):
-    with open(config_path, 'r') as f:
+    with open(config_path, "r") as f:
         config = yaml.safe_load(f)
     if not isinstance(config, dict):
         return None
-    default_agent = config.get('default', 'agent1')
+    default_agent = config.get("default", "agent1")
     agent = config.get(default_agent, {})
     # Check if agent config is empty or missing required fields
-    if not agent or not agent.get('api_key') or not agent.get('api_url'):
+    if not agent or not agent.get("api_key") or not agent.get("api_url"):
         return None
     return agent
 
+
 def send_to_ai_agent(agent, test_name, traceback):
-    api_url = agent.get('api_url')
+    api_url = agent.get("api_url")
     if not api_url:
         return "No api_url provided in agent config."
     headers = {
-        'Authorization': f"Bearer {agent.get('api_key', '')}",
-        'Content-Type': 'application/json'
+        "Authorization": f"Bearer {agent.get('api_key', '')}",
+        "Content-Type": "application/json",
     }
     # Payload for OpenAI chat completions API
     payload = {
-        "model": agent.get('model', 'gpt-4'),
+        "model": agent.get("model", "gpt-4"),
         "messages": [
-            {"role": "system", "content": "You are a helpful test debugging assistant."},
-            {"role": "user", "content": f"Debug this test failure: {test_name}\nTraceback:\n{traceback}"}
-        ]
+            {
+                "role": "system",
+                "content": "You are a helpful test debugging assistant.",
+            },
+            {
+                "role": "user",
+                "content": f"Debug this test failure: {test_name}\nTraceback:\n{traceback}",
+            },
+        ],
     }
     response = requests.post(api_url, json=payload, headers=headers)
     if response.ok:
         try:
-            return response.json().get('choices', [{}])[0].get('message', {}).get('content', '')
+            return (
+                response.json()
+                .get("choices", [{}])[0]
+                .get("message", {})
+                .get("content", "")
+            )
         except Exception:
             return response.text
     # Print error details in red to CLI for visibility
-    RED = '\033[31m'
-    RESET = '\033[0m'
-    print(f"{RED}TESTCATO AI agent error for test '{test_name}': {response.status_code} - {response.text}{RESET}")
+    RED = "\033[31m"
+    RESET = "\033[0m"
+    print(
+        f"{RED}TESTCATO AI agent error for test '{test_name}': {response.status_code} - {response.text}{RESET}"
+    )
     return "AI agent failed to respond."
 
+
 def debug_latest_jsonl():
     # Generate debug JSONL and HTML report after lines, result_dir, and timestamp are defined
-    result_dir = os.path.join(os.getcwd(), 'testcato_result')
-    config_path = os.path.join(os.getcwd(), 'testcato_config.yaml')
+    result_dir = os.path.join(os.getcwd(), "testcato_result")
+    config_path = os.path.join(os.getcwd(), "testcato_config.yaml")
     latest_jsonl = get_latest_jsonl(result_dir)
     if not latest_jsonl:
         print("No test_run JSONL file found.")
         return
     agent = load_agent_config(config_path)
     if not agent:
         # Print warning in yellow in pytest output or console
-        YELLOW = '\033[33m'
-        RESET = '\033[0m'
+        YELLOW = "\033[33m"
+        RESET = "\033[0m"
         warning_msg = f"{YELLOW}WARNING: TESTCATO: No valid AI agent config found. Debugging is disabled.{RESET}"
         import sys
-        tr = getattr(sys, '_pytest_terminalreporter', None)
+
+        tr = getattr(sys, "_pytest_terminalreporter", None)
         if tr:
             tr.write_line(warning_msg)
         else:
             print(warning_msg)
         return
     import json
+
     lines = []
-    with open(latest_jsonl, 'r', encoding='utf-8') as f:
+    with open(latest_jsonl, "r", encoding="utf-8") as f:
         for raw_line in f:
             try:
                 test_data = json.loads(raw_line)
             except Exception:
                 continue
-            test_name = test_data.get('name') or test_data.get('test_name')
-            status = test_data.get('status', 'failed')
-            traceback = test_data.get('traceback')
+            test_name = test_data.get("name") or test_data.get("test_name")
+            status = test_data.get("status", "failed")
+            traceback = test_data.get("traceback")
             debug_result = None
             if traceback:
                 debug_result = send_to_ai_agent(agent, test_name, traceback)
             line = {
                 "test_name": test_name,
                 "status": status,
                 "traceback": traceback,
-                "debug_result": debug_result
+                "debug_result": debug_result,
             }
             lines.append(line)
-    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
-    debug_jsonl_path = os.path.join(result_dir, f'test_debug_{timestamp}.jsonl')
-    with open(debug_jsonl_path, 'w', encoding='utf-8') as f:
+    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+    debug_jsonl_path = os.path.join(result_dir, f"test_debug_{timestamp}.jsonl")
+    with open(debug_jsonl_path, "w", encoding="utf-8") as f:
         for line in lines:
-            f.write(json.dumps(line, ensure_ascii=False) + '\n')
-    GREEN = '\033[32m'
-    RESET = '\033[0m'
+            f.write(json.dumps(line, ensure_ascii=False) + "\n")
+    GREEN = "\033[32m"
+    RESET = "\033[0m"
     print(f"{GREEN}Debug results saved to {debug_jsonl_path}{RESET}")
 
     # Also generate a human-readable HTML report
-    html_path = os.path.join(result_dir, f'test_debug_{timestamp}.html')
-    with open(html_path, 'w', encoding='utf-8') as html:
-        html.write('<html><head><title>Test Debug Report</title></head><body>')
-        html.write('<h1>Test Debug Report</h1>')
+    html_path = os.path.join(result_dir, f"test_debug_{timestamp}.html")
+    with open(html_path, "w", encoding="utf-8") as html:
+        html.write("<html><head><title>Test Debug Report</title></head><body>")
+        html.write("<h1>Test Debug Report</h1>")
         html.write('<table border="1" cellpadding="5" cellspacing="0">')
-        html.write('<tr><th>Test Name</th><th>Status</th><th>Traceback</th><th>Debug Result</th></tr>')
+        html.write(
+            "<tr><th>Test Name</th><th>Status</th><th>Traceback</th><th>Debug Result</th></tr>"
+        )
         for line in lines:
-            html.write('<tr>')
+            html.write("<tr>")
             html.write(f'<td>{line["test_name"]}</td>')
             html.write(f'<td>{line["status"]}</td>')
             html.write(f'<td><pre>{line["traceback"]}</pre></td>')
             html.write(f'<td><pre>{line["debug_result"]}</pre></td>')
-            html.write('</tr>')
-        html.write('</table></body></html>')
+            html.write("</tr>")
+        html.write("</table></body></html>")
     print(f"{GREEN}HTML debug report saved to {html_path}{RESET}")
diff --git a/testcato/categorizer.py b/testcato/categorizer.py
@@ -1,5 +1,6 @@
 # Core logic for categorizing test results
 
+
 class TestCategorizer:
     def __init__(self):
         pass
@@ -10,10 +11,10 @@ def categorize(self, test_results):
         :param test_results: List of test result dicts
         :return: Dict with categories as keys and lists of test names as values
         """
-        categories = {'passed': [], 'failed': [], 'skipped': []}
+        categories = {"passed": [], "failed": [], "skipped": []}
         for result in test_results:
-            status = result.get('status')
-            name = result.get('name')
+            status = result.get("status")
+            name = result.get("name")
             if status in categories:
                 categories[status].append(name)
         return categories
diff --git a/testcato/pytest_testcato.py b/testcato/pytest_testcato.py
@@ -1,52 +1,59 @@
 import re
-import pytest
+
+# import pytest  # Unused import removed
 import os
 import datetime
-## import xml.etree.ElementTree as ET  # No longer needed, switched to JSONL
+
+# import xml.etree.ElementTree as ET  # No longer needed, switched to JSONL
 from .categorizer import TestCategorizer
 
+
 def pytest_addoption(parser):
     parser.addoption(
         "--testcato",
         action="store_true",
         default=False,
-        help="Categorize test results using testcato"
+        help="Categorize test results using testcato",
     )
 
+
 def pytest_configure(config):
     if config.getoption("--testcato"):
         # Add -vvv for maximum verbosity if not already present
         # Pytest config.option.verbose is an int, 0=default, 1=-v, 2=-vv, 3=-vvv
         if getattr(config.option, "verbose", 0) < 3:
             config.option.verbose = 3
 
+
 def pytest_terminal_summary(terminalreporter, exitstatus, config):
     def remove_ansi_codes(text):
-        ansi_escape = re.compile(r'\x1B[@-_][0-?]*[ -/]*[@-~]')
-        return ansi_escape.sub('', text)
+        ansi_escape = re.compile(r"\x1B[@-_][0-?]*[ -/]*[@-~]")
+        return ansi_escape.sub("", text)
+
     if config.getoption("--testcato"):
         results = []
         tracebacks = []
         for report in terminalreporter.getreports("passed"):
-            results.append({'name': report.nodeid, 'status': 'passed'})
+            results.append({"name": report.nodeid, "status": "passed"})
         for report in terminalreporter.getreports("failed"):
-            results.append({'name': report.nodeid, 'status': 'failed'})
-            if hasattr(report, 'longrepr') and report.longrepr:
+            results.append({"name": report.nodeid, "status": "failed"})
+            if hasattr(report, "longrepr") and report.longrepr:
                 tb = remove_ansi_codes(str(report.longrepr))
-                tracebacks.append({'name': report.nodeid, 'traceback': tb})
+                tracebacks.append({"name": report.nodeid, "traceback": tb})
         for report in terminalreporter.getreports("skipped"):
-            results.append({'name': report.nodeid, 'status': 'skipped'})
+            results.append({"name": report.nodeid, "status": "skipped"})
 
         # Save tracebacks to JSON Lines
         if tracebacks:
             import json
-            result_dir = os.path.join(os.getcwd(), 'testcato_result')
+
+            result_dir = os.path.join(os.getcwd(), "testcato_result")
             os.makedirs(result_dir, exist_ok=True)
-            timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
-            jsonl_path = os.path.join(result_dir, f'test_run_{timestamp}.jsonl')
-            with open(jsonl_path, 'w', encoding='utf-8') as f:
+            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+            jsonl_path = os.path.join(result_dir, f"test_run_{timestamp}.jsonl")
+            with open(jsonl_path, "w", encoding="utf-8") as f:
                 for tb in tracebacks:
-                    f.write(json.dumps(tb, ensure_ascii=False) + '\n')
+                    f.write(json.dumps(tb, ensure_ascii=False) + "\n")
             terminalreporter.write_line(f"Tracebacks saved to {jsonl_path}")
 
         categorizer = TestCategorizer()
@@ -57,9 +64,10 @@ def remove_ansi_codes(text):
             for test in tests:
                 terminalreporter.write_line(f"  {test}")
 
-    # Automatically send latest JSONL to AI agent and save debug file
+        # Automatically send latest JSONL to AI agent and save debug file
         try:
             from .ai_agent import debug_latest_jsonl
+
             debug_latest_jsonl()
         except Exception as e:
             terminalreporter.write_line(f"Error sending JSONL to AI agent: {e}")
diff --git a/tests/test_categorizer.py b/tests/test_categorizer.py
diff --git a/tests/test_dummy.py b/tests/test_dummy.py
