hurui200320
bde74c82b5
This MR fixed the pipeline so it can run on forgejo runner. Currently we have two pipelines set up: + One pipeline will run pytest to test the code and collect coverage data, this pipeline will run on all branches and PRs. + One pipeline will build the wheel file and publish it to the company's pypl repo, this pipeline will only run on the master branch and develop branch. This RP also made the following changes to make the pipeline working: + Remove `setup.py` and `requirements.txt`, now the project is fully embraced `pyproject.toml`, which according to ChatGPT, is the most modern way of managing a python project. + Move all test files to `tests` folder, so it won't be packed into the final wheel file and affect the coverage rate. + Change version to `X.Y.Z-dev`, before building the wheel file, the pipeline will automatically add a UTC timestamp after it, so it becomes `X.Y.Z-devYYYYMMDDHHmmSS`. This is added due to the limitation of the pypl repo, unlike maven snapshot versions, you can't push the same version twice with our pypl repository. Reviewed-on: #10 Reviewed-by: Stanislav Hejny <stanislav.hejny@cleverthis.com> Co-authored-by: Rui Hu <rui.hu@cleverthis.com> Co-committed-by: Rui Hu <rui.hu@cleverthis.com>
51 lines
2.2 KiB
Python
51 lines
2.2 KiB
Python
import os
|
|
import tempfile
|
|
from unittest import TestCase
|
|
|
|
from amqp.adapter.file_handler import StreamingFileHandler
|
|
|
|
|
|
class TestFileHandler(TestCase):
|
|
def test_ensure_directory_exists(self):
|
|
"""
|
|
Test function for ensure_directory_exists. Creates a temporary directory
|
|
and files to test the versioning logic.
|
|
"""
|
|
|
|
# Create a temporary directory
|
|
file_handler = StreamingFileHandler()
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Test case 1: File does not exist
|
|
filepath1 = os.path.join(temp_dir, "test1.txt")
|
|
result1 = file_handler.ensure_directory_exists(filepath1)
|
|
assert result1 == filepath1, f"Test Case 1 Failed: {result1} != {filepath1}"
|
|
# Create the file
|
|
with open(filepath1, "w") as f:
|
|
f.write("test")
|
|
|
|
# Test case 2: File exists, should create versioned file
|
|
filepath2 = os.path.join(temp_dir, "test2.txt")
|
|
result2 = file_handler.ensure_directory_exists(filepath2)
|
|
assert result2 == filepath2, f"Test Case 2 Failed: {result2} != {filepath2}"
|
|
with open(filepath2, "w") as f:
|
|
f.write("test")
|
|
result2_v1 = file_handler.ensure_directory_exists(filepath2)
|
|
assert result2_v1 == os.path.join(
|
|
temp_dir, "test2.1.txt"
|
|
), f"Test Case 2 Version 1 Failed: {result2_v1} != {os.path.join(temp_dir, 'test2.1.txt')}"
|
|
|
|
# Create a few versioned files
|
|
with open(os.path.join(temp_dir, "test3.txt"), "w") as f:
|
|
f.write("test")
|
|
with open(os.path.join(temp_dir, "test3.1.txt"), "w") as f:
|
|
f.write("test")
|
|
with open(os.path.join(temp_dir, "test3.2.txt"), "w") as f:
|
|
f.write("test")
|
|
filepath3 = os.path.join(temp_dir, "test3.txt")
|
|
result3_v3 = file_handler.ensure_directory_exists(filepath3)
|
|
assert result3_v3 == os.path.join(
|
|
temp_dir, "test3.3.txt"
|
|
), f"Test Case 3 Version 3 Failed: {result3_v3} != {os.path.join(temp_dir, 'test3.3.txt')}"
|
|
|
|
print("All test cases passed!")
|