Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 11 additions & 3 deletions src/hermes/model/context_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,18 @@ def prepare_step(self, step: str, *depends: str) -> None:
self._current_step.append(step)

def finalize_step(self, step: str) -> None:
current_step = self._current_step.pop()
if current_step != step:
raise ValueError("Cannot end step %s while in %s.", step, self._current_step[-1])
if len(self._current_step) < 1:
raise ValueError("There is no step to end.")
if self._current_step[-1] != step:
raise ValueError(f"Cannot end step {step} while in {self._current_step[-1]}.")
self._current_step.pop()

def __getitem__(self, source_name: str) -> HermesCache:
if len(self._current_step) < 1:
raise HermesContexError("Prepare a step first.")
subdir = self.cache_dir / self._current_step[-1] / source_name
return HermesCache(subdir)


class HermesContexError(Exception):
pass
3 changes: 3 additions & 0 deletions test/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
# SPDX-FileCopyrightText: 2025 German Aerospace Center (DLR)
#
# SPDX-License-Identifier: Apache-2.0
2 changes: 1 addition & 1 deletion test/hermes_test/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
# SPDX-FileCopyrightText: 2022 German Aerospace Center (DLR)
# SPDX-FileCopyrightText: 2025 German Aerospace Center (DLR)
#
# SPDX-License-Identifier: Apache-2.0
38 changes: 0 additions & 38 deletions test/hermes_test/model/test_base_context.py

This file was deleted.

87 changes: 87 additions & 0 deletions test/hermes_test/model/test_context_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
# SPDX-FileCopyrightText: 2025 German Aerospace Center (DLR)
#
# SPDX-License-Identifier: Apache-2.0

# SPDX-FileContributor: Michael Meinel
# SPDX-FileContributor: Sophie Kernchen
import pytest
from pathlib import Path

from hermes.model.context_manager import HermesContext, HermesCache, HermesContexError


def test_context_hermes_dir_default():
ctx = HermesContext()
assert ctx.cache_dir == Path('./.hermes').absolute()


def test_context_hermes_dir_custom():
ctx = HermesContext(Path('spam'))
assert ctx.cache_dir == Path('spam') / '.hermes'


def test_get_context_default():
ctx = HermesContext()
ctx.prepare_step("ham")
assert ctx["spam"]._cache_dir == Path('./.hermes/ham/spam').absolute()


def test_context_get_error():
ctx = HermesContext()
ctx.prepare_step("ham")
ctx.finalize_step("ham")
with pytest.raises(HermesContexError, match="Prepare a step first."):
ctx["spam"]._cache_dir == Path('./.hermes/spam').absolute()


def test_finalize_context_one_item():
ctx = HermesContext()
ctx.prepare_step("ham")
ctx.finalize_step("ham")
assert ctx._current_step == []


def test_finalize_context():
ctx = HermesContext()
ctx.prepare_step("ham")
ctx.prepare_step("spam")
ctx.finalize_step("spam")
assert ctx["spam"]._cache_dir == Path('./.hermes/ham/spam').absolute()
assert ctx._current_step == ["ham"]


def test_finalize_context_error_no_item():
ctx = HermesContext()
with pytest.raises(ValueError, match="There is no step to end."):
ctx.finalize_step("spam")


def test_finalize_context_error():
ctx = HermesContext()
ctx.prepare_step("ham")
with pytest.raises(ValueError, match="Cannot end step spam while in ham."):
ctx.finalize_step("spam")


def test_cache(tmpdir):
ctx = HermesContext(Path(tmpdir))
ctx.prepare_step("ham")
with ctx["spam"] as c:
c["bacon"] = {"data": "goose", "num": 2}

path = tmpdir / Path('.hermes') / 'ham' / 'spam' / 'bacon.json'
assert path.exists()
assert path.read() == '{"data": "goose", "num": 2}'
assert c._cached_data["bacon"] == {"data": "goose", "num": 2}


def test_hermes_cache_set(tmpdir):
cache = HermesCache(Path(tmpdir))
cache["bacon"] = "eggs"
assert cache._cached_data == {"bacon": "eggs"}


def test_hermes_cache_get(tmpdir):
cache = HermesCache(Path(tmpdir))
cache._cached_data = {"bacon": "eggs"}
assert cache["bacon"] == "eggs"
Loading