Modification to test settings (#80)

* changed setting for testing

* changed utils.py

---------

Co-authored-by: John Calderon <john.calderon@centml.ai>

Author: johncalesp

test/config.json

@@ -0,0 +1,5 @@
+def TestConfig():
+    config = dict()
+    config["model_names_from_examples"] = ["resnet", "nanogpt"]
+
+    return config

test/config_params.py

@@ -2,7 +2,8 @@
 import random
 import deepview_profile.db.database as database
 
-
+LOWER_BOUND_RAND_INT = 1
+UPPER_BOUND_RAND_INT = 10
 class MockDatabaseInterface(database.DatabaseInterface):
     def __del__(self):
         if os.path.exists("test.sqlite"):
@@ -37,30 +38,45 @@ def test_invalid_entry_wrong_types(self):
         )
 
     def test_adding_valid_entry(self):
-        params = ["entry_point", random.random(), random.random(), random.randint()]
+        params = [
+            "entry_point",
+            random.random(),
+            random.random(),
+            random.randint(LOWER_BOUND_RAND_INT, UPPER_BOUND_RAND_INT),
+        ]
         self.energy_table_interface.add_entry(params)
         query_result = self.test_database.connection.execute(
-            "SELECT * FROM ENERGY;"
+            "SELECT * FROM ENERGY ORDER BY ts DESC;"
         ).fetchone()
         # params is passed in by reference so it have the timestamp in it
         assert query_result == tuple(params)
 
     # add 10 valid entries and get top 3
     def test_get_latest_n_entries_of_entry_point(self):
         for _ in range(10):
-            params = ["entry_point", random.random(), random.random(), random.randint()]
+            params = [
+                "entry_point",
+                random.random(),
+                random.random(),
+                random.randint(LOWER_BOUND_RAND_INT, UPPER_BOUND_RAND_INT),
+            ]
             self.energy_table_interface.add_entry(params)
         for _ in range(20):
             params = [
                 "other_entry_point",
                 random.random(),
                 random.random(),
-                random.randint(),
+                random.randint(LOWER_BOUND_RAND_INT, UPPER_BOUND_RAND_INT),
             ]
             self.energy_table_interface.add_entry(params)
         entries = []
         for _ in range(3):
-            params = ["entry_point", random.random(), random.random(), random.randint()]
+            params = [
+                "entry_point",
+                random.random(),
+                random.random(),
+                random.randint(LOWER_BOUND_RAND_INT, UPPER_BOUND_RAND_INT),
+            ]
             entries.insert(0, params)
             self.energy_table_interface.add_entry(params)
         latest_n_entries = (
@@ -69,4 +85,4 @@ def test_get_latest_n_entries_of_entry_point(self):
             )
         )
         entries = [tuple(entry) for entry in entries]
-        assert entries == latest_n_entries
+        assert entries == latest_n_entries

test/test_database.py

@@ -1,31 +1,81 @@
-
 import pytest
-import json
-from utils import SkylineSession, BackendContext
+import pickle
+from utils import DeepviewSession, BackendContext
+from google.protobuf.json_format import MessageToDict
+from config_params import TestConfig
+import os
+
+REPS = 2
+NUM_EXPECTED_MESSAGES = 6
+
+
+def get_config_name():
+    import pkg_resources
 
-with open("config.json", "r") as fp:
-    config = json.load(fp)
+    package_versions = {p.key: p.version for p in pkg_resources.working_set}
+    return package_versions
+
+
+config = TestConfig()
 
 tests = list()
-for entry_point in config["entry_points"]:
-    tests.append((config["skyline_bin"], entry_point))
+for model_name in config["model_names_from_examples"]:
+    dir_path = os.path.join(
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+        "examples",
+        model_name,
+    )
+    tests.append((model_name, dir_path))
+
 
-@pytest.mark.parametrize("skyline_bin,entry_point", tests)
-def test_entry_point(skyline_bin, entry_point):
+@pytest.mark.parametrize("test_name, entry_point", tests)
+def test_entry_point(test_name, entry_point):
     print(f"Testing {entry_point}")
-    context = BackendContext(skyline_bin, entry_point)
+
+    # create new folder
+    folder = (
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/tests_results"
+    )
+    os.makedirs(folder, exist_ok=True)
+
+    stdout_fd = open(os.path.join(folder, f"{test_name}_interactive_output.log"), "w")
+    stderr_fd = open(os.path.join(folder, f"{test_name}_interactive_w_debug_output.log"), "w")
+    context = BackendContext(entry_point, stdout_fd=stdout_fd, stderr_fd=stderr_fd)
     context.spawn_process()
 
-    sess = SkylineSession()
-    while context.state == 0:
-        pass
-    sess.connect("localhost", 60120)
-    sess.send_initialize_request()
-    sess.send_analysis_request()
-    while len(sess.received_messages) < 4:
-        pass
+    analysis_messages = list()
 
-    sess.cleanup()
-    context.terminate()
+    for reps in range(REPS):
+        sess = DeepviewSession()
+        while context.state == 0:
+            pass
+        sess.connect("localhost", 60120)
+        sess.send_initialize_request(entry_point)
+        sess.send_analysis_request()
+        while (
+            context.alive()
+            and sess.alive()
+            and len(sess.received_messages) < NUM_EXPECTED_MESSAGES
+        ):
+            pass
 
-    assert(len(sess.received_messages) == 4)
+        sess.cleanup()
+        analysis_messages.extend(sess.received_messages)
+
+        assert len(sess.received_messages) == NUM_EXPECTED_MESSAGES, (
+            f"Run {reps}: Expected to receive {NUM_EXPECTED_MESSAGES} got "
+            f"{len(sess.received_messages)} (did the process terminate prematurely?)"
+        )
+
+    context.terminate()
+    # create folder to store files
+    # flush contents to files
+    with open(os.path.join(folder, f"{test_name}_analysis.pkl"), "wb") as fp:
+        pickle.dump(list(map(MessageToDict, analysis_messages)), fp)
+    # write package versions
+    package_dict = get_config_name()
+    with open(os.path.join(folder, "package-list.txt"), "w") as f:
+        for k, v in package_dict.items():
+            f.write(f"{k}={v}\n")
+    stdout_fd.close()
+    stderr_fd.close()

test/test_driver.py

@@ -1,4 +1,5 @@
 import os
+import select
 import socket
 import struct
 import subprocess
@@ -17,27 +18,47 @@ def stream_monitor(stream, callback=None):
 
 def socket_monitor(socket, callback=None):
     try:
-        while True:
-            msg_len = struct.unpack(">I", socket.recv(4))[0]
-            msg = socket.recv(msg_len)
-            print(f"Received message of length {msg_len}")
-            callback(msg)
-    except OSError:
+        while socket:
+            ret = select.select([socket], [], [], 1)
+            if len(ret[0]):
+                msg_len = struct.unpack(">I", socket.recv(4))[0]
+                buf = None
+                while socket and msg_len:
+                    msg = socket.recv(msg_len)
+                    if len(msg) == 0: break
+                    buf = buf + msg if buf else msg
+                    msg_len -= len(msg)
+
+                callback(buf)
+    except (ValueError, OSError):
+        # terminate whenever either recv() or select() fails.
+        # - if the socket is closed, then recv() will raise OSError as it attempts to read
+        #   from a closed file descriptor.
+        # - if the socket is closed, then select() will receive an invalid (i.e. negative) file descriptor
+        #   and will raise a ValueError
         print(f"Closing listener for socket {socket}")
 
 
 class BackendContext:
-    def __init__(self, skyline_bin, entry_point):
+    def __init__(self, entry_point, stdout_fd=None, stderr_fd=None):
         self.process = None
-        self.skyline_bin = skyline_bin
         self.entry_point = entry_point
         self.state = 0
+        self.stdout_fd = stdout_fd
+        self.stderr_fd = stderr_fd
 
     def on_message_stdout(self, message):
-        message = message.decode("ascii").rstrip()
+        # message = message.decode("ascii").rstrip()
+        message = message.decode("ascii")
+        if self.stdout_fd:
+            self.stdout_fd.write(message)
 
     def on_message_stderr(self, message):
-        message = message.decode("ascii").rstrip()
+        message = message.decode("ascii")
+        if self.stderr_fd:
+            self.stderr_fd.write(message)
+
+        message = message.rstrip()
         print("stderr", message)
         if "DeepView interactive profiling session started!" in message:
             self.state = 1
@@ -46,8 +67,7 @@ def spawn_process(self):
         # DeepView expects the entry_point filename to be relative
         working_dir = os.path.dirname(self.entry_point)
         entry_filename = os.path.basename(self.entry_point)
-        launch_command = [self.skyline_bin, "interactive", entry_filename]
-
+        launch_command = ["python", "-m", "deepview_profile", "interactive", "--debug"]
         # Launch backend + listener threads for stdout and stderr
         self.process = subprocess.Popen(
             launch_command,
@@ -64,6 +84,9 @@ def spawn_process(self):
         )
         self.stderr_thread.start()
 
+    def alive(self):
+        return self.process and self.process.poll() is None
+
     def join(self):
         self.process.wait()
 
@@ -73,7 +96,7 @@ def terminate(self):
         self.stderr_thread.join()
 
 
-class SkylineSession:
+class DeepviewSession:
     def __init__(self):
         self.seq_num = 0
         self.received_messages = []
@@ -103,9 +126,11 @@ def send_message(self, message):
         self.socket.sendall(length_buffer)
         self.socket.sendall(buf)
 
-    def send_initialize_request(self):
+    def send_initialize_request(self, project_root):
         request = innpv_pb2.InitializeRequest()
         request.protocol_version = 5
+        request.project_root = project_root
+        request.entry_point = "entry_point.py"
         self.send_message(request)
 
     def send_analysis_request(self):
@@ -116,12 +141,17 @@ def send_analysis_request(self):
     def handle_message(self, message):
         from_server = innpv_pb2.FromServer()
         from_server.ParseFromString(message)
-        # print("From Server:")
-        # print(from_server)
+        print("From Server:")
+        print(from_server)
 
         self.received_messages.append(from_server)
         print(f"new message. total: {len(self.received_messages)}")
 
+    def alive(self):
+        # makes sure that the last message is not error
+        return (not self.received_messages) or \
+            not self.received_messages[-1].HasField("analysis_error")
+
     def cleanup(self):
         # Closing the socket should cause the listener thread to die
         self.socket.close()