-
-
Notifications
You must be signed in to change notification settings - Fork 153
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
remove unittest class and fix failing test case for receiving recordi…
…ng, and clean up share.py
- Loading branch information
1 parent
786e063
commit 644c83a
Showing
2 changed files
with
86 additions
and
70 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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) |