From 644c83a02a5ab0274801b39212c08eb2f9b36009 Mon Sep 17 00:00:00 2001 From: Mustafa Abdulrahman Date: Sat, 22 Jul 2023 18:34:37 -0400 Subject: [PATCH] remove unittest class and fix failing test case for receiving recording, and clean up share.py --- openadapt/share.py | 16 +++- tests/openadapt/test_share.py | 140 ++++++++++++++++++---------------- 2 files changed, 86 insertions(+), 70 deletions(-) diff --git a/openadapt/share.py b/openadapt/share.py index 3a8a0d1cb..8d7162cbe 100644 --- a/openadapt/share.py +++ b/openadapt/share.py @@ -122,7 +122,6 @@ def receive_recording(wormhole_code: str) -> None: except subprocess.CalledProcessError as exc: logger.exception(exc) - return finally: # Delete the zip file after sending or in case of exception if os.path.exists(zip_path): @@ -131,10 +130,21 @@ def receive_recording(wormhole_code: str) -> None: def visualize_recording(db_name: str) -> None: + """Visualize a recording from a SQLite database. + + This function loads the specified db file containing a recording and + visualizes the data using the 'visualize.main' function. + + Args: + db_name (str): The name of the SQLite database containing the recording. + + Raises: + sqlalchemy.exc.OperationalError: If there is an error accessing the database. + """ if db_name == "openadapt.db": - recording_path = config.ROOT_DIRPATH / db_name + recording_path = os.path.join(config.ROOT_DIRPATH, db_name) else: - recording_path = config.RECORDING_DIRECTORY_PATH / db_name + recording_path = os.path.join(config.RECORDING_DIRECTORY_PATH, db_name) recording_url = f"sqlite:///{recording_path}" engine = create_engine(recording_url, future=True) diff --git a/tests/openadapt/test_share.py b/tests/openadapt/test_share.py index 3a310cb8d..92de03cb6 100644 --- a/tests/openadapt/test_share.py +++ b/tests/openadapt/test_share.py @@ -1,78 +1,84 @@ +"""Module to test scrub.py""" + +from unittest.mock import patch +from zipfile import ZIP_DEFLATED, ZipFile +import os import subprocess import tempfile -import unittest -import os -from unittest.mock import patch -from openadapt import share +from openadapt import config, share + + +def test_export_recording_to_folder() -> None: + # Create a temporary recording database file + recording_id = 1 + recording_db_path = "temp.db" + with open(recording_db_path, "w") as f: + f.write("Recording data") + + # Mock the crud.export_recording() function to return the temporary file path + with patch("openadapt.share.db.export_recording", return_value=recording_db_path): + zip_file_path = share.export_recording_to_folder(recording_id) + + assert zip_file_path is not None + assert os.path.exists(zip_file_path) + + # Assert that the file is removed after calling export_recording_to_folder + assert not os.path.exists(recording_db_path), "Temporary file was not removed." + + +def test_send_file() -> None: + # Create a temporary file + with tempfile.NamedTemporaryFile(delete=False) as temp_file: + file_path = temp_file.name + temp_file.write(b"File data") + + # Mock the subprocess.run() function to avoid executing the command + with patch("openadapt.share.subprocess.run") as mock_run: + share.send_file(file_path) + + # Verify that the command is called with the correct arguments + mock_run.assert_called_once_with(["wormhole", "send", file_path], check=True) + + # Clean up the temporary file + os.remove(file_path) + + +def test_send_recording() -> None: + # Mock the export_recording_to_folder() function to return a zip file path + with patch( + "openadapt.share.export_recording_to_folder", + return_value=str(config.RECORDING_DIRECTORY_PATH / "recording_1.zip"), + ): + # Mock the send_file() function to avoid sending the file + with patch("openadapt.share.send_file"): + share.send_recording(1) -class ShareTestCase(unittest.TestCase): - """ - A test case for the share module. + # Verify that export_recording_to_folder() and send_file() are called + assert share.export_recording_to_folder.called + assert share.send_file.called - This test case verifies the functionality of the share module, - including exporting a recording to a folder, sending a file, - sending a recording, and receiving a recording. - """ + # Verify that the temporary zip file is deleted after the test + assert not os.path.exists(config.RECORDING_DIRECTORY_PATH / "recording_1.zip") - def test_export_recording_to_folder(self): - # Create a temporary recording database file - recording_id = 1 - recording_db_path = "temp.db" - with open(recording_db_path, "w") as f: - f.write("Recording data") - # Mock the crud.export_recording() function to return the temporary file path - with patch( - "openadapt.share.db.export_recording", return_value=recording_db_path - ): - zip_file_path = share.export_recording_to_folder(recording_id) +# Test receive_recording function (mock the subprocess.run function) +def test_receive_recording() -> None: + # Create a temporary zip file + temp_zip_path = str(config.RECORDING_DIRECTORY_PATH / "recording.zip") + with ZipFile(temp_zip_path, "w", ZIP_DEFLATED): + pass # Using the 'pass' statement creates an empty zip file - self.assertIsNotNone(zip_file_path) - self.assertTrue(os.path.exists(zip_file_path)) + # Mock the subprocess.run() function to avoid executing the command + with patch("subprocess.run"): + # Simulate receiving a recording with a wormhole code + wormhole_code = "test_wormhole_code" + share.receive_recording(wormhole_code) - # Assert that the file is removed after calling export_recording_to_folder - self.assertFalse( - os.path.exists(recording_db_path), "Temporary file was not removed." + # Verify that the command is called with the correct arguments + subprocess.run.assert_called_once_with( + ["wormhole", "receive", "-o", temp_zip_path, wormhole_code], check=True ) - def test_send_file(self): - # Create a temporary file - with tempfile.NamedTemporaryFile(delete=False) as temp_file: - file_path = temp_file.name - temp_file.write(b"File data") - - # Mock the subprocess.run() function to avoid executing the command - with patch("openadapt.share.subprocess.run") as mock_run: - share.send_file(file_path) - - # Verify that the command is called with the correct arguments - mock_run.assert_called_once_with( - ["wormhole", "send", file_path], check=True - ) - - # Clean up the temporary file - os.remove(file_path) - - def test_send_recording(self): - # Mock the export_recording_to_folder() function to return a zip file path - with patch( - "openadapt.share.export_recording_to_folder", return_value="temp.zip" - ): - # Mock the send_file() function to avoid sending the file - with patch("openadapt.share.send_file"): - share.send_recording(1) - - # Verify that export_recording_to_folder() and send_file() are called - self.assertTrue(share.export_recording_to_folder.called) - self.assertTrue(share.send_file.called) - - def test_receive_recording(self): - # Mock the subprocess.run() function to avoid executing the command - with patch("openadapt.share.subprocess.run"): - share.receive_recording("wormhole_code") - - # Verify that the command is called with the correct arguments - subprocess.run.assert_called_once_with( - ["wormhole", "receive", "wormhole_code"], check=True - ) + # Verify that the zip file has been deleted + assert not os.path.exists(temp_zip_path)