Update TestCase.load() to in place

- Remove TestCase.purge_optional since this now happens automatically.
- TestCase.load() now only loads a single test case.
- Break up test case loading in to smaller more testable parts.

grizzly/adapter/adapter.py

@@ -16,11 +16,11 @@ class AdapterError(Exception):
 
 
 class Adapter(metaclass=ABCMeta):
-    """An Adapter is an interface between Grizzly and a fuzzer. A subclass must
+    """An Adapter is the interface between Grizzly and a fuzzer. A subclass must
     be created in order to add support for additional fuzzers. The Adapter is
     responsible for handling input/output data and executing the fuzzer.
     It is expected that any processes launched or file created on file system
-    in the adapter will also be cleaned up in the adapter.
+    by the adapter will also be cleaned up by the adapter.
     NOTE: Some methods must not be overloaded doing so will prevent Grizzly from
     operating correctly.
 
@@ -34,8 +34,6 @@ class Adapter(metaclass=ABCMeta):
                          remaining to process.
     """
 
-    # Only report test cases with served content.
-    IGNORE_UNSERVED = True
     # Maximum iterations between Target relaunches (<1 use default)
     RELAUNCH = 0
     # Maximum execution time per test (used as minimum timeout). The iteration is

grizzly/adapter/test_adapter.py

@@ -54,24 +54,24 @@ def test_adapter_03(tmp_path):
     # empty path
     assert not any(SimpleAdapter.scan_path(str(tmp_path)))
     # missing path
-    assert not any(SimpleAdapter.scan_path(str(tmp_path / "none")))
+    assert not any(SimpleAdapter.scan_path(tmp_path / "none"))
     # path to file
     file1 = tmp_path / "test1.txt"
     file1.touch()
-    found = tuple(SimpleAdapter.scan_path(str(file1)))
+    found = tuple(SimpleAdapter.scan_path(file1))
     assert str(file1) in found
     assert len(found) == 1
     # path to directory
-    assert len(tuple(SimpleAdapter.scan_path(str(tmp_path)))) == 1
+    assert len(tuple(SimpleAdapter.scan_path(tmp_path))) == 1
     # path to directory (w/ ignored)
     (tmp_path / ".ignored").touch()
     nested = tmp_path / "nested"
     nested.mkdir()
     file2 = nested / "test2.bin"
     file2.touch()
-    assert len(tuple(SimpleAdapter.scan_path(str(tmp_path)))) == 1
+    assert len(tuple(SimpleAdapter.scan_path(tmp_path))) == 1
     # path to directory (recursive)
-    found = tuple(SimpleAdapter.scan_path(str(tmp_path), recursive=True))
+    found = tuple(SimpleAdapter.scan_path(tmp_path, recursive=True))
     assert str(file1) in found
     assert str(file2) in found
     assert len(found) == 2

grizzly/common/runner.py

@@ -277,6 +277,9 @@ def run(
             # overwrite instead of replace 'grz_next_test' for consistency
             server_map.set_redirect("grz_next_test", "grz_empty", required=True)
             server_map.set_dynamic_response("grz_empty", lambda _: b"", required=True)
+        # clear optional contents from test case
+        # it will be repopulated with served contests
+        testcase.clear_optional()
         # serve the test case
         serve_start = time()
         server_status, served = self._server.serve_path(
@@ -293,13 +296,13 @@ def run(
             attempted=testcase.entry_point in served,
             timeout=server_status == Served.TIMEOUT,
         )
-        # add all include files that were served to test case
-        if server_map.include:
-            existing = set(testcase.contents)
-            for url, local_file in served.items():
-                if url in existing:
-                    continue
-                testcase.add_from_file(local_file, file_name=url, copy=True)
+        # add all files that were served (includes, etc...) to test
+        existing = set(testcase.required)
+        for url, local_file in served.items():
+            if url in existing:
+                continue
+            # use copy here so include files are copied
+            testcase.add_from_file(local_file, file_name=url, copy=True)
         # record use of https in testcase
         testcase.https = self._server.scheme == "https"
         if result.timeout: