Skip to content
Snippets Groups Projects
Commit f316f0eb authored by Manon Blanco's avatar Manon Blanco
Browse files

Comply with ruff's PTH rule

parent af111533
No related branches found
No related tags found
1 merge request!454Comply with ruff's PTH rule
Pipeline #144884 passed
......@@ -27,18 +27,19 @@ def decompress_zst_archive(compressed_archive: Path) -> tuple[int, Path]:
"""
dctx = zstandard.ZstdDecompressor()
archive_fd, archive_path = tempfile.mkstemp(suffix=".tar")
archive_path = Path(archive_path)
logger.debug(f"Uncompressing file to {archive_path}")
try:
with open(compressed_archive, "rb") as compressed, open(
archive_path, "wb"
with compressed_archive.open("rb") as compressed, archive_path.open(
"wb"
) as decompressed:
dctx.copy_stream(compressed, decompressed)
logger.debug(f"Successfully uncompressed archive {compressed_archive}")
except zstandard.ZstdError as e:
raise Exception(f"Couldn't uncompressed archive: {e}") from e
return archive_fd, Path(archive_path)
return archive_fd, archive_path
def extract_tar_archive(archive_path: Path, destination: Path):
......
......@@ -56,6 +56,8 @@ select = [
"SIM",
# flake8-pytest-style
"PT",
# flake8-use-pathlib
"PTH",
]
[tool.ruff.per-file-ignores]
......
......@@ -332,20 +332,19 @@ def mock_elements_worker_with_cache(monkeypatch, mock_cache_db, _mock_worker_run
@pytest.fixture()
def fake_page_element():
with open(FIXTURES_DIR / "page_element.json") as f:
return json.load(f)
return json.loads((FIXTURES_DIR / "page_element.json").read_text())
@pytest.fixture()
def fake_ufcn_worker_version():
with open(FIXTURES_DIR / "ufcn_line_historical_worker_version.json") as f:
return json.load(f)
return json.loads(
(FIXTURES_DIR / "ufcn_line_historical_worker_version.json").read_text()
)
@pytest.fixture()
def fake_transcriptions_small():
with open(FIXTURES_DIR / "line_transcriptions_small.json") as f:
return json.load(f)
return json.loads((FIXTURES_DIR / "line_transcriptions_small.json").read_text())
@pytest.fixture()
......
......@@ -30,22 +30,20 @@ def test_init(tmp_path):
def test_create_tables_existing_table(tmp_path):
db_path = f"{tmp_path}/db.sqlite"
db_path = tmp_path / "db.sqlite"
# Create the tables once…
init_cache_db(db_path)
create_tables()
db.close()
with open(db_path, "rb") as before_file:
before = before_file.read()
before = db_path.read_bytes()
# Create them again
init_cache_db(db_path)
create_tables()
with open(db_path, "rb") as after_file:
after = after_file.read()
after = db_path.read_bytes()
assert before == after, "Existing table structure was modified"
......
import json
import os
import sys
import tempfile
from pathlib import Path
from uuid import UUID
import pytest
......@@ -12,48 +12,50 @@ from arkindex_worker.worker import ElementsWorker
@pytest.mark.usefixtures("_mock_worker_run_api")
def test_cli_default(monkeypatch):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump(
path = Path(path)
path.write_text(
json.dumps(
[
{"id": "volumeid", "type": "volume"},
{"id": "pageid", "type": "page"},
{"id": "actid", "type": "act"},
{"id": "surfaceid", "type": "surface"},
],
f,
)
)
monkeypatch.setenv("TASK_ELEMENTS", path)
monkeypatch.setattr(sys, "argv", ["worker"])
worker = ElementsWorker()
worker.configure()
assert worker.args.elements_list.name == path
assert worker.args.elements_list.name == str(path)
assert not worker.args.element
os.unlink(path)
path.unlink()
@pytest.mark.usefixtures("_mock_worker_run_api")
def test_cli_arg_elements_list_given(mocker):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump(
path = Path(path)
path.write_text(
json.dumps(
[
{"id": "volumeid", "type": "volume"},
{"id": "pageid", "type": "page"},
{"id": "actid", "type": "act"},
{"id": "surfaceid", "type": "surface"},
],
f,
)
)
mocker.patch.object(sys, "argv", ["worker", "--elements-list", path])
mocker.patch.object(sys, "argv", ["worker", "--elements-list", str(path)])
worker = ElementsWorker()
worker.configure()
assert worker.args.elements_list.name == path
assert worker.args.elements_list.name == str(path)
assert not worker.args.element
os.unlink(path)
path.unlink()
def test_cli_arg_element_one_given_not_uuid(mocker):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment