Added wrapper functions for refreshing latest version, quota usage.

Author: LTLA

src/pygobbler/__init__.py

@@ -35,3 +35,5 @@
 from .clone_version import clone_version
 from .service_info import service_info
 from .version_path import version_path
+from .refresh_latest import refresh_latest
+from .refresh_usage import refresh_usage

src/pygobbler/refresh_latest.py

@@ -0,0 +1,31 @@
+from typing import Optional
+from . import _utils as ut
+
+
+def refresh_latest(project: str, asset: str, staging: str, url:str) -> Optional[str]:
+    """
+    Recompute the latest version of a project's asset. This is useful on rare
+    occasions where multiple simultaneous uploads cause the latest version to
+    be slightly out of sync.
+
+    Args:
+        project:
+            Name of the project.
+
+        asset:
+            Name of the asset.
+
+        staging:
+            Path to the staging directory.
+
+        url:
+            URL of the gobbler REST API.
+
+    Returns:
+        Latest version of the project, or None if there is no non-probational version.
+    """
+    resp = ut.dump_request(staging, url, "refresh_latest", { "project": project, "asset": asset })
+    if "version" in resp:
+        return resp["version"]
+    else:
+        return None

src/pygobbler/refresh_usage.py

@@ -0,0 +1,24 @@
+from . import _utils as ut
+
+
+def refresh_usage(project: str, staging: str, url:str) -> int:
+    """
+    Recompute the quota usage of a project. This is useful on rare occasions
+    where multiple simultaneous uploads cause the usage calculations to be out
+    of sync.
+
+    Args:
+        project:
+            Name of the project.
+
+        staging:
+            Path to the staging directory.
+
+        url:
+            URL of the gobbler REST API.
+
+    Returns:
+        Total quota usage of this project, in bytes.
+    """
+    resp = ut.dump_request(staging, url, "refresh_usage", { "project": project })
+    return resp["total"]

tests/test_refresh_latest.py

@@ -0,0 +1,33 @@
+import pygobbler as pyg
+import pytest
+import os
+import time
+import tempfile
+
+
+@pytest.fixture(scope="module")
+def setup():
+    _, staging, registry, url = pyg.start_gobbler()
+
+    pyg.remove_project("test", staging=staging, url=url)
+    pyg.create_project("test", staging=staging, url=url)
+
+    tmp = tempfile.mkdtemp()
+    pyg.upload_directory(project="test", asset="latest", version="v1", directory=tmp, staging=staging, url=url)
+    time.sleep(1.1)
+    pyg.upload_directory(project="test", asset="latest", version="v2", directory=tmp, staging=staging, url=url)
+    time.sleep(1.1)
+    pyg.upload_directory(project="test", asset="latest", version="v3", directory=tmp, staging=staging, url=url)
+
+
+def test_refresh_latest(setup):
+    _, staging, registry, url = pyg.start_gobbler()
+
+    assert pyg.fetch_latest("test", "latest", registry=registry, url=url) == "v3"
+
+    os.unlink(os.path.join(registry, "test", "latest", "..latest"))
+    assert pyg.fetch_latest("test", "latest", registry=registry, url=url) is None 
+
+    v = pyg.refresh_latest("test", "latest", staging=staging, url=url)
+    assert v == "v3"
+    assert pyg.fetch_latest("test", "latest", registry=registry, url=url) == "v3"

tests/test_refresh_usage.py

@@ -0,0 +1,36 @@
+import pygobbler as pyg
+import pytest
+import os
+import tempfile
+
+
+@pytest.fixture(scope="module")
+def setup():
+    _, staging, registry, url = pyg.start_gobbler()
+
+    pyg.remove_project("test-usage", staging=staging, url=url)
+    pyg.create_project("test-usage", staging=staging, url=url)
+
+    tmp = tempfile.mkdtemp()
+    with open(os.path.join(tmp, "blah.txt"), "w") as f:
+        f.write("BAR")
+    os.mkdir(os.path.join(tmp, "foo"))
+    with open(os.path.join(tmp, "foo", "bar.txt"), "w") as f:
+        f.write("1 2 3 4 5 6 7 8 9 10")
+
+    pyg.upload_directory(project="test-usage", asset="quota", version="v1", directory=tmp, staging=staging, url=url)
+
+
+def test_refresh_usage(setup):
+    _, staging, registry, url = pyg.start_gobbler()
+
+    assert pyg.fetch_usage("test-usage", registry=registry, url=url) > 0
+
+    with open(os.path.join(registry, "test-usage", "..usage"), "w") as handle:
+        handle.write('{ "total": 0 }')
+    assert pyg.fetch_usage("test-usage", registry=registry, url=url) == 0
+
+    # Fixing the project usage.
+    out = pyg.refresh_usage("test-usage", staging=staging, url=url)
+    assert out > 0
+    assert pyg.fetch_usage("test-usage", registry=registry, url=url) == out