fix: timestamp format in lock files

Author: fyhertz

src/ops2deb/lockfile.py

@@ -25,8 +25,8 @@ class LockFile(BaseModel):
     __root__: list[LockEntry]
 
 
-def get_iso_utc_datetime() -> str:
-    return datetime.now(tz=timezone.utc).isoformat()[:-13] + "Z"
+def get_iso_utc_datetime() -> datetime:
+    return datetime.now(tz=timezone.utc).replace(microsecond=0)
 
 
 class Lock:
@@ -84,18 +84,20 @@ def save(self) -> None:
         if not self._entries or self._tainted is False:
             return
 
-        entries = {k: entry.dict() for k, entry in self._entries.items()}
-
         # make sure all added urls since lock was created have the same timestamp
         # and make sure this timestamp is when save() was called
         now = get_iso_utc_datetime()
         for new_url in self._new_urls:
-            if (entry := entries.get(new_url, None)) is not None:
-                entry["timestamp"] = now
+            if (entry := self._entries.get(new_url, None)) is not None:
+                entry.timestamp = now
+
+        # sort lockfile entries by urls
+        entries = [entry.dict() for entry in self._entries.values()]
+        sorted_entries = sorted(entries, key=itemgetter("url"))
 
         with self.lock_file_path.open("w") as output:
             yaml.dump(
-                sorted(entries.values(), key=itemgetter("url")),
+                sorted_entries,
                 output,
                 Dumper=PrettyYAMLDumper,
                 default_flow_style=False,

src/ops2deb/updater.py

@@ -146,7 +146,9 @@ async def __call__(self, blueprint: Blueprint) -> str:
         if Version.isvalid(version) and Version.isvalid(blueprint.version):
             version = str(max(Version.parse(version), Version.parse(blueprint.version)))
         if await self._try_version(blueprint, version) is False:
-            raise Ops2debUpdaterError("Failed to determine latest release URL")
+            raise Ops2debUpdaterError(
+                f"Failed to determine latest release URL (latest tag is {tag_name})"
+            )
         return version
 
 

src/ops2deb/utils.py

@@ -1,39 +1,19 @@
 import os
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Iterator, Tuple, TypeVar
+from typing import Any, Iterator
 
 import yaml
 from ruamel.yaml.emitter import Emitter
 
 from ops2deb import logger
-from ops2deb.exceptions import Ops2debError
 
 
 def log_and_raise(exception: Exception) -> None:
     logger.error(str(exception))
     raise exception
 
 
-T = TypeVar("T")
-U = TypeVar("U")
-
-
-def separate_results_from_errors(
-    results_and_errors: dict[U, T | Exception]
-) -> Tuple[dict[U, T], dict[U, Ops2debError]]:
-    results: dict[U, T] = {}
-    errors: dict[U, Ops2debError] = {}
-    for key, value in results_and_errors.items():
-        if isinstance(value, Ops2debError):
-            errors[key] = value
-        elif isinstance(value, Exception):
-            raise value
-        else:
-            results[key] = value
-    return results, errors
-
-
 @contextmanager
 def working_directory(path: Path) -> Iterator[None]:
     origin = Path().absolute()
@@ -45,6 +25,9 @@ def working_directory(path: Path) -> Iterator[None]:
 
 
 class PrettyYAMLDumper(yaml.dumper.SafeDumper):
+    def ignore_aliases(self, data: Any) -> bool:
+        return True
+
     def expect_block_sequence(self) -> None:
         self.increase_indent(flow=False, indentless=False)
         self.state = self.expect_first_block_sequence_item

tests/test_lockfile.py

@@ -1,74 +1,175 @@
+from dataclasses import dataclass
+from datetime import datetime, timezone
 from unittest.mock import patch
 
 import pytest
 
 from ops2deb.exceptions import Ops2debLockFileError
-from ops2deb.fetcher import FetchResult
 from ops2deb.lockfile import Lock
 
 
-def test__init__should_create_empty_lock_when_lockfile_does_not_exist(lockfile_path):
+@dataclass
+class UrlAndHash:
+    url: str
+    sha256: str
+
+
+def test_init__creates_an_empty_lock_when_lockfile_does_not_exist(lockfile_path):
+    # Given
     lock = Lock(lockfile_path)
+
+    # Then
     assert lock._entries == {}
 
 
-def test__init__should_raise_when_lockfile_path_is_a_directory(tmp_path):
+def test_init__raises_when_lockfile_path_is_a_directory(tmp_path):
+    # When
     with pytest.raises(Ops2debLockFileError) as error:
         Lock(tmp_path)
+
+    # Then
     assert error.match("Path points to a directory")
 
 
-def test__init__should_raise_when_lockfile_path_contains_invalid_yaml(lockfile_path):
+def test_init__raises_when_lockfile_path_contains_invalid_yaml(lockfile_path):
+    # Given
     lockfile_path.write_text("@£¢±")
+
+    # When
     with pytest.raises(Ops2debLockFileError) as error:
         Lock(lockfile_path)
+
+    # Then
     assert error.match("Invalid YAML file")
 
 
-def test__init__should_raise_when_lockfile_cannot_be_parsed_with_pydantic(lockfile_path):
+def test_init__raises_when_lockfile_cannot_be_parsed_with_pydantic(lockfile_path):
+    # Given
     lockfile_path.write_text("1")
+
+    # When
     with pytest.raises(Ops2debLockFileError) as error:
         Lock(lockfile_path)
+
+    # Then
     assert error.match("Invalid lockfile")
 
 
-def test_sha256__should_raise_when_url_is_not_in_cache(lockfile_path):
+def test_sha256__raises_when_url_is_not_in_cache(lockfile_path):
+    # Given
     url = "http://tests.com/file.tar.gz"
+
+    # When
     with pytest.raises(Ops2debLockFileError) as error:
         Lock(lockfile_path).sha256(url)
+
+    # Then
     assert error.match(f"Unknown hash for url {url}, please run ops2deb lock")
 
 
 def test_save__should_not_create_a_file_when_lock_is_empty(lockfile_path):
-    Lock(lockfile_path).save()
+    # Given
+    lock = Lock(lockfile_path)
+
+    # When
+    lock.save()
+
+    # Then
     assert lockfile_path.exists() is False
 
 
-def test_save__should_produce_a_lockfile_that_contains_added_entries(
-    lockfile_path, tmp_path
-):
+def test_save__produces_a_lockfile_that_contains_added_entries(lockfile_path):
+    # Given
     lock = Lock(lockfile_path)
-    lock.add([FetchResult("http://tests.com/file.tar.gz", "deadbeef", tmp_path, None)])
+    lock.add([UrlAndHash("http://tests.com/file.tar.gz", "deadbeef")])
+
+    # When
     lock.save()
-    lock = Lock(lockfile_path)
-    assert lock.sha256("http://tests.com/file.tar.gz") == "deadbeef"
+
+    # Then
+    assert Lock(lockfile_path).sha256("http://tests.com/file.tar.gz") == "deadbeef"
 
 
 @patch("yaml.dump")
 def test_save__should_not_write_file_when_no_entry_have_been_added_nor_removed(
     mock_dump, lockfile_path
 ):
+    # Given
     lock = Lock(lockfile_path)
+
+    # When
     lock.save()
+
+    # Then
     mock_dump.assert_not_called()
 
 
-def test_save__set_the_same_timestamp_to_added_entries(lockfile_path, tmp_path):
+def test_save__sets_the_same_timestamp_to_added_entries(lockfile_path):
+    # Given
     lock = Lock(lockfile_path)
-    lock.add([FetchResult("http://tests.com/file1.tar.gz", "deadbeef", tmp_path, None)])
-    lock.add([FetchResult("http://tests.com/file2.tar.gz", "deadbeef", tmp_path, None)])
+    lock.add([UrlAndHash("http://tests.com/file1.tar.gz", "deadbeef")])
+    lock.add([UrlAndHash("http://tests.com/file2.tar.gz", "deadbeef")])
+
+    # When
     lock.save()
+
+    # Then
     lock = Lock(lockfile_path)
     timestamp_1 = lock.timestamp("http://tests.com/file1.tar.gz")
     timestamp_2 = lock.timestamp("http://tests.com/file2.tar.gz")
     assert timestamp_1 == timestamp_2
+
+
+@patch("ops2deb.lockfile.datetime")
+def test_save__should_not_include_microseconds_in_timestamps(
+    mock_datetime, lockfile_path
+):
+    # Given
+    mock_datetime.now.return_value = datetime(
+        2023, 3, 4, 0, 22, 14, 1234, tzinfo=timezone.utc
+    )
+    lock = Lock(lockfile_path)
+    lock.add([UrlAndHash("http://tests.com/file1.tar.gz", "deadbeef")])
+
+    # When
+    lock.save()
+
+    # Then
+    assert "2023-03-04 00:22:14+00:00" in lockfile_path.read_text()
+
+
+@patch("ops2deb.lockfile.datetime")
+def test_save__should_be_idempotent(mock_datetime, lockfile_path):
+    # Given
+    mock_datetime.now.return_value = datetime(2023, 3, 4, 0, 22, 14, tzinfo=timezone.utc)
+
+    # When
+    lock = Lock(lockfile_path)
+    lock.add([UrlAndHash("http://tests.com/file1.tar.gz", "deadbeef")])
+    lock.add([UrlAndHash("http://tests.com/file2.tar.gz", "deadbeef")])
+    lock.save()
+    lockfile_content_0 = lockfile_path.read_text()
+    Lock(lockfile_path).save()
+    lockfile_content_1 = lockfile_path.read_text()
+    print(lockfile_content_0)
+
+    # Then
+    assert lockfile_content_0 == lockfile_content_1
+
+
+@patch("ops2deb.lockfile.datetime")
+def test_save__should_not_use_yaml_anchors_in_timestamps(mock_datetime, lockfile_path):
+    # happens when you reference an object multiple time in a YAML document and when the
+    # pyyaml dumper is not configured to "ignore aliases"
+
+    # Given
+    mock_datetime.now.return_value = datetime(2023, 3, 4, 0, 22, 14, tzinfo=timezone.utc)
+
+    # When
+    lock = Lock(lockfile_path)
+    lock.add([UrlAndHash("http://tests.com/file1.tar.gz", "deadbeef")])
+    lock.add([UrlAndHash("http://tests.com/file2.tar.gz", "deadbeef")])
+    lock.save()
+
+    # Then
+    assert "timestamp: &" not in lockfile_path.read_text()

tests/test_ops2deb.py

@@ -683,11 +683,12 @@ def test_update__creates_a_summary_of_updated_blueprints_when_called_with_output
     )
 
     # Then
-    summary = (
-        "Updated great-app from 1.0.0 to 1.1.1\n"
-        "Updated super-app from 1.0.0 to 1.1.1\n"
-    )
-    assert summary_path.read_text() == summary
+    summary_lines_set = {
+        "",
+        "Updated great-app from 1.0.0 to 1.1.1",
+        "Updated super-app from 1.0.0 to 1.1.1",
+    }
+    assert set(summary_path.read_text().split("\n")) == summary_lines_set
 
 
 def test_update__creates_empty_summary_when_called_with_output_file_and_configuration_is_up_to_date(  # noqa: E501

tests/test_utils.py

@@ -1,38 +1,17 @@
 import os
 from pathlib import Path
 
-import pytest
+from ops2deb.utils import working_directory
 
-from ops2deb.exceptions import Ops2debError
-from ops2deb.utils import separate_results_from_errors, working_directory
 
-
-def test_separate_results_from_errors_should_separate_results_from_ops2deb_exceptions():
-    error = Ops2debError("An error")
-    success = "success"
-    test = {0: error, 1: success, 2: error, 4: success}
-    results, errors = separate_results_from_errors(test)
-    assert errors == {0: error, 2: error}
-    assert results == {1: success, 4: success}
-
-
-def test_separate_results_from_errors_should_raise_when_exception_is_not_an_ops2deb_error():  # noqa: E501
-    error = RuntimeError("An error")
-    test = {0: error}
-    with pytest.raises(RuntimeError):
-        separate_results_from_errors(test)
-
-
-def test_working_directory__should_set_current_working_directory_within_context(tmp_path):
+def test_working_directory__sets_current_working_directory_within_context(tmp_path):
     origin = Path().absolute()
     with working_directory(tmp_path):
         assert Path(os.getcwd()) != origin
         assert Path(os.getcwd()) == tmp_path
 
 
-def test_working_directory__should_restore_current_directory_when_context_is_left(
-    tmp_path,
-):
+def test_working_directory__restores_current_directory_when_context_is_left(tmp_path):
     origin = Path().absolute()
     with working_directory(tmp_path):
         pass