-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' into opendal-comments
- Loading branch information
Showing
21 changed files
with
237 additions
and
150 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
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
13 changes: 13 additions & 0 deletions
13
packages/jumpstarter-driver-dutlink/jumpstarter_driver_dutlink/conftest.py
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 |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| import pytest | ||
| import usb | ||
|
|
||
|
|
||
| def pytest_runtest_call(item): | ||
| try: | ||
| item.runtest() | ||
| except FileNotFoundError: | ||
| pytest.skip("dutlink not available") | ||
| except usb.core.USBError: | ||
| pytest.skip("USB not available") | ||
| except usb.core.NoBackendError: | ||
| pytest.skip("No USB backend") |
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
76 changes: 49 additions & 27 deletions
76
packages/jumpstarter-driver-dutlink/jumpstarter_driver_dutlink/driver_test.py
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,42 +1,64 @@ | ||
| import pytest | ||
| import usb | ||
| from time import sleep | ||
|
|
||
| from jumpstarter_driver_network.adapters import PexpectAdapter | ||
|
|
||
| from jumpstarter_driver_dutlink.driver import Dutlink | ||
| from jumpstarter_driver_dutlink.driver import Dutlink, DutlinkPower, DutlinkSerial, DutlinkStorageMux | ||
|
|
||
| from jumpstarter.common.utils import serve | ||
|
|
||
| STORAGE_DEVICE = "/dev/null" # MANUAL: replace with path to block device | ||
|
|
||
| def test_drivers_dutlink(): | ||
| try: | ||
| instance = Dutlink( | ||
| storage_device="/dev/null", | ||
| ) | ||
| except FileNotFoundError: | ||
| pytest.skip("dutlink not available") | ||
| except usb.core.USBError: | ||
| pytest.skip("USB not available") | ||
| except usb.core.NoBackendError: | ||
| pytest.skip("No USB backend") | ||
|
|
||
| def power_test(power): | ||
| power.on() # MANUAL: led DUT_ON should be turned on | ||
| sleep(1) | ||
| assert next(power.read()).current != 0 | ||
| power.off() # MANUAL: led DUT_ON should be turned off | ||
|
|
||
|
|
||
| def storage_test(storage): | ||
| storage.write_local_file("/dev/null") | ||
|
|
||
|
|
||
| def serial_test(serial): | ||
| with PexpectAdapter(client=serial) as expect: | ||
| expect.send("\x02" * 5) | ||
|
|
||
| expect.send("about\r\n") | ||
| expect.expect("Jumpstarter test-harness") | ||
|
|
||
| expect.send("console\r\n") | ||
| expect.expect("Entering console mode") | ||
|
|
||
| expect.send("hello") | ||
| expect.expect("hello") | ||
|
|
||
|
|
||
| def test_drivers_dutlink_power(): | ||
| instance = DutlinkPower() | ||
|
|
||
| with serve(instance) as client: | ||
| with PexpectAdapter(client=client.console) as expect: | ||
| expect.send("\x02" * 5) | ||
| power_test(client) | ||
|
|
||
| expect.send("about\r\n") | ||
| expect.expect("Jumpstarter test-harness") | ||
|
|
||
| expect.send("console\r\n") | ||
| expect.expect("Entering console mode") | ||
| def test_drivers_dutlink_storage_mux(): | ||
| instance = DutlinkStorageMux(storage_device=STORAGE_DEVICE) | ||
|
|
||
| client.power.off() | ||
| with serve(instance) as client: | ||
| storage_test(client) | ||
|
|
||
| client.storage.write_local_file("/dev/null") | ||
| client.storage.dut() | ||
|
|
||
| client.power.on() | ||
| def test_drivers_dutlink_serial(): | ||
| instance = DutlinkSerial() # MANUAL: connect tx to rx | ||
|
|
||
| expect.send("\x02" * 5) | ||
| expect.expect("Exiting console mode") | ||
| with serve(instance) as client: | ||
| serial_test(client) | ||
|
|
||
| client.power.off() | ||
|
|
||
| def test_drivers_dutlink(): | ||
| instance = Dutlink(storage_device=STORAGE_DEVICE) | ||
|
|
||
| with serve(instance) as client: | ||
| power_test(client.power) | ||
| storage_test(client.storage) | ||
| serial_test(client.console) |
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
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
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,51 +1,37 @@ | ||
| import logging | ||
| import time | ||
|
|
||
| import pytest | ||
| from jumpstarter_driver_tftp.driver import FileNotFound, TftpError | ||
| from jumpstarter_testing.pytest import JumpstarterTest | ||
|
|
||
| log = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class TestResource(JumpstarterTest): | ||
| filter_labels = {"board": "rpi4"} | ||
|
|
||
| @pytest.fixture() | ||
| def test_tftp_upload(self, client): | ||
| def setup_tftp(self, client): | ||
| # Move the setup code to a fixture | ||
| client.tftp.start() | ||
| yield client | ||
| client.tftp.stop() | ||
|
|
||
| def test_tftp_operations(self, setup_tftp): | ||
| client = setup_tftp | ||
| test_file = "test.bin" | ||
|
|
||
| # Create test file | ||
| with open(test_file, "wb") as f: | ||
| f.write(b"Hello from TFTP streaming test!") | ||
|
|
||
| try: | ||
| client.tftp.start() | ||
| print("TFTP server started") | ||
|
|
||
| time.sleep(1) | ||
|
|
||
| test_file = "test.bin" | ||
| with open(test_file, "wb") as f: | ||
| f.write(b"Hello from TFTP streaming test!") | ||
|
|
||
| try: | ||
| client.tftp.put_local_file(test_file) | ||
| print(f"Successfully uploaded {test_file}") | ||
|
|
||
| files = client.tftp.list_files() | ||
| print(f"Files in TFTP root: {files}") | ||
|
|
||
| if test_file in files: | ||
| client.tftp.delete_file(test_file) | ||
| print(f"Successfully deleted {test_file}") | ||
| else: | ||
| print(f"Warning: {test_file} not found in TFTP root") | ||
|
|
||
| except TftpError as e: | ||
| print(f"TFTP operation failed: {e}") | ||
| except FileNotFound as e: | ||
| print(f"File not found: {e}") | ||
|
|
||
| except Exception as e: | ||
| print(f"Error: {e}") | ||
| finally: | ||
| try: | ||
| client.tftp.stop() | ||
| print("TFTP server stopped") | ||
| except Exception as e: | ||
| print(f"Error stopping server: {e}") | ||
| # Test upload | ||
| client.tftp.put_local_file(test_file) | ||
| assert test_file in client.tftp.list_files() | ||
|
|
||
| # Test delete | ||
| client.tftp.delete_file(test_file) | ||
| assert test_file not in client.tftp.list_files() | ||
|
|
||
| except (TftpError, FileNotFound) as e: | ||
| pytest.fail(f"Test failed: {e}") |
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 |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| CHUNK_SIZE = 1024 * 1024 * 4 # 4MB |
Oops, something went wrong.