Feature: Pytest In A Subprocess

.gitignore

@@ -1,3 +1,6 @@
+# Project-specific rules
+build/
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

TestRunner/GenericTestRunner.py

@@ -0,0 +1,44 @@
+import pytest
+import subprocess
+from abc import ABCMeta, abstractmethod
+from pytest_plugins import ResultsCollector, SessionStartPlugin
+
+
+class GenericTestRunner(metaclass=ABCMeta):
+    @abstractmethod
+    def run(self, *args, **kwargs) -> (int, str): pass
+
+
+class SubProcessTestRunner(GenericTestRunner):
+    code_dir: str
+    test_dir: str
+
+    def __init__(self, _code, _test) -> None:
+        self.code_dir = _code
+        self.test_dir = _test
+
+    def run(self, *args, **kwargs) -> (int, str):
+        # TODO: check that code_dir and test_dir exist
+        proc = subprocess.run(["pytest", self.test_dir], capture_output=True, universal_newlines=True)
+        return proc.returncode, proc.stdout
+
+
+class InlineTestRunner(GenericTestRunner):
+
+    def __init__(self, _code, _test) -> None:
+        self.code_dir = _code
+        self.test_dir = _test
+
+    def run(self, *args, **kwargs) -> (int, str):
+        collector = ResultsCollector()
+        setup = SessionStartPlugin()
+        pytest.main(args=["-k", "ExampleClass"], plugins=[collector, setup])
+        _out = ""
+
+        if collector.exitcode > 0:
+            for report in collector.reports:
+                _out += f"{report.outcome.upper()} {report.nodeid} ... Outcome: - {report.longrepr.reprcrash.message}"
+                _out += "\n"
+                _out += report.longreprtext
+                _out += "\n"
+        return collector.exitcode, _out

generated/test_class.py

@@ -1 +0,0 @@
-# THIS FILE INTENTIONALLY LEFT BLANK

main.py

@@ -1,27 +1,18 @@
 import os
 import typer
 
-# for the test runner
-import time
-import pytest
-# ------------------
-
 import langroid as lr
 from langroid.utils.configuration import set_global, Settings
 from langroid.utils.logging import setup_colored_logging
 
-# import the empty generated code
-import generated.test_class
-# ------------------
+from TestRunner.GenericTestRunner import GenericTestRunner, InlineTestRunner, SubProcessTestRunner
 
 app = typer.Typer()
 setup_colored_logging()
 
 
-def generate_first_attempt() -> None:
-    # code_prompt = typer.Prompt("Describe what kind of code you want.")
-    # class_skeleton: str = ""
-    with open(os.path.join(".", "assets", "test_class.py"), "r") as f:
+def generate_first_attempt(class_skeleton: str) -> None:
+    with open(class_skeleton, "r") as f:
         class_skeleton = f.read()
 
     cfg = lr.ChatAgentConfig(
@@ -43,65 +34,7 @@ def generate_first_attempt() -> None:
         _out.write(response.content)
 
 
-class ResultsCollector:
-    def __init__(self):
-        self.reports = []
-        self.collected = 0
-        self.exitcode = 0
-        self.passed = 0
-        self.failed = 0
-        self.xfailed = 0
-        self.skipped = 0
-        self.total_duration = 0
-
-    @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_makereport(self, item, call):
-        outcome = yield
-        report = outcome.get_result()
-        if report.when == 'call':
-            self.reports.append(report)
-
-    def pytest_collection_modifyitems(self, items):
-        self.collected = len(items)
-
-    def pytest_terminal_summary(self, terminalreporter, exitstatus):
-        self.exitcode = exitstatus
-        self.passed = len(terminalreporter.stats.get('passed', []))
-        self.failed = len(terminalreporter.stats.get('failed', []))
-        self.xfailed = len(terminalreporter.stats.get('xfailed', []))
-        self.skipped = len(terminalreporter.stats.get('skipped', []))
-
-        self.total_duration = time.time() - terminalreporter._sessionstarttime
-
-
-class SessionStartPlugin:
-    """
-    The goal of this plugin is to allow us to run pytest multiple times
-    and have it pick up the changes we generate in `generated/test_class.py`
-    """
-    def pytest_sessionstart(self):
-        if globals().get('generated', None) is not None:
-            import importlib
-            print("Reloading generated.test_class module...")
-            importlib.reload(generated.test_class)
-
-
-def get_test_results() -> str:
-    collector = ResultsCollector()
-    setup = SessionStartPlugin()
-    pytest.main(args=["-k", "ExampleClass"], plugins=[collector, setup])
-    _out = ""
-
-    if collector.exitcode > 0:
-        for report in collector.reports:
-            _out += f"{report.outcome.upper()} {report.nodeid} ... Outcome: - {report.longrepr.reprcrash.message}"
-            _out += "\n"
-            _out += report.longreprtext
-            _out += "\n"
-    return collector.exitcode, _out
-
-
-def generate_next_attempt(test_results: str) -> None:
+def generate_next_attempt(test_results: str, test_results_insights: str) -> None:
     cfg = lr.ChatAgentConfig(
         llm=lr.language_models.OpenAIGPTConfig(
             chat_model="ollama/llama3:latest",
@@ -120,34 +53,83 @@ def generate_next_attempt(test_results: str) -> None:
             {code_snippet}
             Here are the test results:
             {test_results}
+            In addition, you may consider these insights about the test results when coming up with your solution:
+            {test_results_insights}
             Update the code so that the tests will pass.
             Your output MUST contain all the same classes and methods as the input code.
             Do NOT add any other methods or commentary.
             Your response should be ONLY the python code.
             Do not say 'here is the python code'
             Do not surround your response with quotes or backticks.
+            Your response should NEVER start or end with ```
             Your output MUST be valid, runnable python code and NOTHING else.
         """
     response = agent.llm_response(prompt)
     with open(os.path.join(".", "generated", "test_class.py"), "w") as _out:
         _out.write(response.content)
 
 
-def chat() -> None:
-    generate_first_attempt()
-    for _ in range(5):
-        test_exit_code, test_results = get_test_results()
-        if test_exit_code == 1:
-            generate_next_attempt(test_results)
+def interpret_test_results(results: str) -> str:
+    cfg = lr.ChatAgentConfig(
+        llm=lr.language_models.OpenAIGPTConfig(
+            chat_model="ollama/llama3:latest",
+            chat_context_length=8000,
+        ),
+        vecdb=None
+    )
+    agent = lr.ChatAgent(cfg)
+    prompt = f"""
+        You are an expert at interpreting the results of unit tests, and providing insight into what they mean.
+        You should be descriptive about what variables are incorrect, and in what way.
+        You should include information about which methods should be modified, and in what way.
+        You should generally not provide code.
+        Please provide insights about the following test results:
+        {results}
+    """
+    response = agent.llm_response(prompt)
+    return response.content
+
+
+def teardown() -> None:
+    codegen_path = os.path.join(".", "generated")
+    build_path = os.path.join(".", "build")
+    if not os.path.exists(build_path):
+        os.makedirs(build_path)
+    with open(os.path.join(codegen_path, "test_class.py"), "r+") as generated_file:
+        with open(os.path.join(build_path, "test_class.py"), "w+") as _out:
+            _out.write(generated_file.read())
+            generated_file.truncate(0)
+
+
+def chat(class_skeleton: str, test_dir: str, test_runner: GenericTestRunner, max_epochs: int=5) -> None:
+    generate_first_attempt(class_skeleton)
+    solved = False
+    for _ in range(max_epochs):
+        # test_exit_code, test_results = get_test_results()
+        test_exit_code, test_results = test_runner.run()
+        print(test_results)
+        if test_exit_code == 0:
+            solved = True
+            print("Done!")
+            break
+        elif test_exit_code == 1:
+            results_insights = interpret_test_results(test_results)
+            generate_next_attempt(test_results, results_insights)
         else:
+            solved = True
+            print("There is some problem with the test suite itself.")
             break
-    print("Done! All tests are passing, or there is some problem with the test suite itself.")
+    teardown()
+    if not solved:
+        print(f"Reached the end of epoch {max_epochs} without finding a solution :(")
 
 @app.command()
 def main(
     debug: bool = typer.Option(False, "--debug", "-d", help="debug mode"),
     no_stream: bool = typer.Option(False, "--nostream", "-ns", help="no streaming"),
     nocache: bool = typer.Option(False, "--nocache", "-nc", help="don't use cache"),
+    class_skeleton: str = typer.Option(None, "--class-skeleton", "-c", help="You must provide a class skeleton."),
+    test_dir: str = typer.Option(os.path.join(".", "test"), "--test-dir", "-t", help=""),
 ) -> None:
     set_global(
         Settings(
@@ -156,7 +138,15 @@ def main(
             stream=not no_stream,
         )
     )
-    chat()
+    assert os.path.isfile(class_skeleton), f"The class skeleton file provided does not exist! Got {class_skeleton}"
+    assert os.path.exists(test_dir), f"The test-dir provided does not exist! Got {test_dir}"
+
+    tr: GenericTestRunner = SubProcessTestRunner("", test_dir)
+    chat(
+        class_skeleton=class_skeleton,
+        test_dir=test_dir,
+        test_runner=tr
+    )
 
 
 if __name__ == "__main__":

pytest_plugins.py

@@ -0,0 +1,49 @@
+import pytest
+import time
+try:
+    import generated.test_class
+except ImportError:
+    pass  # Since this is a generated file, it sometimes doesn't exist.
+
+
+class ResultsCollector:
+    def __init__(self):
+        self.reports = []
+        self.collected = 0
+        self.exitcode = 0
+        self.passed = 0
+        self.failed = 0
+        self.xfailed = 0
+        self.skipped = 0
+        self.total_duration = 0
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_makereport(self, item, call):
+        outcome = yield
+        report = outcome.get_result()
+        if report.when == 'call':
+            self.reports.append(report)
+
+    def pytest_collection_modifyitems(self, items):
+        self.collected = len(items)
+
+    def pytest_terminal_summary(self, terminalreporter, exitstatus):
+        self.exitcode = exitstatus
+        self.passed = len(terminalreporter.stats.get('passed', []))
+        self.failed = len(terminalreporter.stats.get('failed', []))
+        self.xfailed = len(terminalreporter.stats.get('xfailed', []))
+        self.skipped = len(terminalreporter.stats.get('skipped', []))
+
+        self.total_duration = time.time() - terminalreporter._sessionstarttime
+
+
+class SessionStartPlugin:
+    """
+    The goal of this plugin is to allow us to run pytest multiple times
+    and have it pick up the changes we generate in `generated/test_class.py`
+    """
+    def pytest_sessionstart(self):
+        if globals().get('generated', None) is not None:
+            import importlib
+            print("Reloading generated.test_class module...")
+            importlib.reload(generated.test_class)

test/test_ExampleClass.py

@@ -1,5 +1,5 @@
 import pytest
-import generated
+from generated import test_class
 
 
 @pytest.mark.parametrize("num1, num2, expected1, expected2", [
@@ -11,6 +11,6 @@
 def test_init(num1, num2, expected1, expected2):
     # Let's make a test with a tricky expectation.
     # When we initialize TestClass, we should subtract 1 from x and y.
-    instance = generated.test_class.ExampleClass(num1, num2)
+    instance = test_class.ExampleClass(num1, num2)
     assert instance.x == expected1
     assert instance.y == expected2