Adds more tests.

bitia/checksumdir.py

@@ -25,7 +25,7 @@ HASH_FUNCS = {
 
 def dirhash(
     dirname: Path,
-    hashfunc: str = "md5",
+    hashfunc: str = "sha256",
     excluded_files: T.List[Path] = [],
     ignore_hidden: bool = False,
     followlinks: bool = False,
@@ -55,7 +55,8 @@ def dirhash(
     return _reduce_hash(hashvalues, hash_func)
 
 
-def filehash(filepath: Path, hashfunc: str = "md5"):
+def filehash(filepath: Path, hashfunc: str = "sha256"):
+    """Compute checksum of a file"""
     hasher = HASH_FUNCS[hashfunc]()
     blocksize = 64 * 1024
 

bitia/common.py

 from pathlib import Path
-import hashlib
 
 from bitia.logger import logger
+from bitia.checksumdir import filehash
 
 
 def sha256sum(infile: Path) -> str:
-    """Compute sha256sum of a file.
-
-    Credit
-    ------
-    Thanks https://stackoverflow.com/a/44873382/1805129
-
-    """
-    h = hashlib.sha256()
-    b = bytearray(128 * 1024)
-    mv = memoryview(b)
-    with infile.open("rb", buffering=0) as handler:
-        while n := handler.readinto(mv):
-            h.update(mv[:n])
-    return h.hexdigest()
+    """Compute sha256sum of a file."""
+    return filehash(infile, "sha256")
 
 
 def dir_info(user_dir: Path) -> dict:

bitia/pipeline.py

@@ -5,8 +5,10 @@ import uuid
 import shutil
 import tempfile
 import zipfile
+import json
 from pathlib import Path
-from datetime import datetime
+
+#  from datetime import datetime
 
 import validators
 import bitia.config as bconfig
@@ -15,24 +17,7 @@ from bitia.logger import logger
 from bitia.checksumdir import dirhash
 
 
-class Singleon(type):
-    _instances: T.Dict[T.Any, T.Any] = {}
-
-    def __call__(cls, zipfile, user_input=None, **kwargs):
-        if cls not in cls._instances:
-            cls._instances[cls] = super(Singleon, cls).__call__(
-                zipfile, user_input, **kwargs
-            )
-        else:
-            prev_instance = cls._instances[cls]
-            if prev_instance.user_input != user_input:
-                cls._instances[cls] = super(Singleon, cls).__call__(
-                    zipfile, user_input, **kwargs
-                )
-        return cls._instances[cls]
-
-
-class Pipeline(metaclass=Singleon):
+class Pipeline:
     def __init__(self, zipfile: Path, user_input=None, **kwargs):
         self.zipfile: Path = zipfile
         self.hash: str = sha256sum(zipfile)
@@ -41,6 +26,13 @@ class Pipeline(metaclass=Singleon):
     def __eq__(self, other):
         return self.checksum == other.checksum and self.size == other.size
 
+    def __repr__(self) -> str:
+        return json.dumps(
+            dict(file=str(self.zipfile), checksum=self.checksum, size=self.size),
+            sort_keys=True,
+            indent=2,
+        )
+
     @property
     def size(self) -> int:
         return self.zipfile.stat().st_size
@@ -50,23 +42,16 @@ class Pipeline(metaclass=Singleon):
         return self.hash
 
     @classmethod
-    def from_command(
-        cls, cmd: str, add_timestamp: bool = True, add_uuid1: bool = True
-    ) -> "Pipeline":
-        """Create a pipeline from user input.
-
-        Returns
-        -------
-        The directory in which pipeline was created.
+    def from_command(cls, cmd: str, comment: str = "") -> "Pipeline":
+        """Create a pipeline from user command. The semantics is a bit different
+        for this.
         """
         pipeline_dir = Path(tempfile.mkdtemp(prefix="bitia_"))
         pipeline_file = pipeline_dir / bconfig.BITIA_MAIN_SCRIPT_NAME
         lines: T.List[str] = []
         lines.append("#!/bin/sh")
-        if add_timestamp:
-            lines.append(f"# timestamp={datetime.now().isoformat()}")
-        if add_uuid1:
-            lines.append(f"# uuid={uuid.uuid1()}")
+        assert "\n" not in comment, "An eol char is found in comment"
+        lines.append(f"# {comment}")
         lines.append(f"{cmd}")
         pipeline_file.write_text("\n".join(lines))
         logger.debug("Wrote pipeline %s", pipeline_file.read_text())

tests/test_methods.py

+import tempfile
+from pathlib import Path
+
+from bitia.common import sha256sum
+
+
+def test_filehash():
+    tempdir = tempfile.gettempdir()
+    t1 = "ls -ltrh /"
+    t2 = "ls -ltrh"
+    f1 = Path(tempdir) / "a.txt"
+    f2 = Path(tempdir) / "b.txt"
+    f1.write_text(t1)
+    f2.write_text(t1)
+    assert sha256sum(f1) == sha256sum(f2)
+
+    f2.write_text(t2)
+    assert sha256sum(f1) != sha256sum(f2)
+
+    f1.write_text(t2)
+    assert sha256sum(f1) == sha256sum(f2)

tests/test_pipeline.py

@@ -4,7 +4,7 @@ from bitia.pipeline import Pipeline
 
 
 def test_pipeline_sanity():
-    uinput = "ls -ltrh"
+    uinput = "ls -l"
     pipeline = Pipeline.from_command(uinput)
     assert pipeline
     assert pipeline.user_input == uinput
@@ -17,9 +17,10 @@ def test_pipeline_singleton():
     u2 = "ls -ltrh /"
     p1 = Pipeline.from_command(u1)
     p2 = Pipeline.from_command(u1)
+    assert p1 == p2
     p3 = Pipeline.from_command(u2)
     p4 = Pipeline.from_command(u2)
-    assert p1 == p2
     assert p3 == p4
+
     assert p2 != p3
     assert p1 != p4