Repository: mandarons/icloud-drive-docker
Pull request #126 (2023-02-09 16:28:42)

Added application usage
Track application usage to determine if I should still continue to maintain this project.


Patch:
diff --git a/.github/workflows/official-release-to-docker-hub.yml b/.github/workflows/official-release-to-docker-hub.yml
index db7a0ebf..70b49365 100644
--- a/.github/workflows/official-release-to-docker-hub.yml
+++ b/.github/workflows/official-release-to-docker-hub.yml
@@ -56,6 +56,10 @@ jobs:
         with:
           context: ./
           file: ./Dockerfile
+          build-args: |
+            NEW_INSTALLATION_ENDPOINT=${{ secrets.NEW_INSTALLATION_ENDPOINT }}
+            NEW_HEARTBEAT_ENDPOINT=${{ secrets.NEW_HEARTBEAT_ENDPOINT }}
+            APP_VERSION=${GITHUB_REF#refs/tags/v}
           push: true
           platforms: linux/386,linux/amd64, linux/arm64, linux/arm/7
           tags: ${{ steps.prep.outputs.tags }}
diff --git a/Dockerfile b/Dockerfile
index 9326fa4c..b76c0366 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -14,4 +14,7 @@ RUN apk update && apk add libmagic
 ENV PATH="/app/venv/bin/:$PATH"
 ENV PYTHONPATH /app
 COPY . /app/
+ARG APP_VERSION=dev
+ARG NEW_INSTALLATION_ENDPOINT=dev
+ARG NEW_HEARTBEAT_ENDPOINT=dev
 CMD ["python", "-u", "./src/main.py"]
diff --git a/src/sync.py b/src/sync.py
index 8f7b3159..ad5773d1 100644
--- a/src/sync.py
+++ b/src/sync.py
@@ -16,6 +16,7 @@ from src import (
     sync_drive,
     sync_photos,
 )
+from src.usage import alive
 
 
 def get_api_instance(
@@ -52,6 +53,7 @@ def sync():
     sleep_for = 10
     while True:
         config = read_config()
+        alive(config=config)
         username = config_parser.get_username(config=config)
         if username:
             try:
diff --git a/src/usage.py b/src/usage.py
new file mode 100644
index 00000000..1330819c
--- /dev/null
+++ b/src/usage.py
@@ -0,0 +1,106 @@
+"""To record usage of the app."""
+import json
+import os
+
+import requests
+
+from src.config_parser import prepare_root_destination
+
+CACHE_FILE_NAME = ".data"
+NEW_INSTALLATION_ENDPOINT = os.environ.get("NEW_INSTALLATION_ENDPOINT", None)
+NEW_HEARTBEAT_ENDPOINT = os.environ.get("NEW_HEARTBEAT_ENDPOINT", None)
+APP_NAME = "icloud-drive-docker"
+APP_VERSION = os.environ.get("APP_VERSION", "dev")
+NEW_INSTALLATION_DATA = {"appName": APP_NAME, "appVersion": APP_VERSION}
+
+
+def init_cache(config):
+    """Initialize the cache file."""
+    root_destination_path = prepare_root_destination(config=config)
+    cache_file_path = os.path.join(root_destination_path, CACHE_FILE_NAME)
+    return cache_file_path
+
+
+def load_cache(file_path: str):
+    """Load the cache file."""
+    data = {}
+    if os.path.isfile(file_path):
+        with open(file_path) as f:
+            data = json.load(f)
+    else:
+        save_cache(file_path=file_path, data={})
+    return data
+
+
+def save_cache(file_path: str, data: object):
+    """Save data to the cache file."""
+    with open(file_path, "w") as f:
+        json.dump(data, f)
+    return True
+
+
+def post_new_installation(data, endpoint=NEW_INSTALLATION_ENDPOINT):
+    """Post new installation to server."""
+    try:
+        response = requests.post(endpoint, data)
+        if response.ok:
+            data = response.json()
+            return data["id"]
+    except Exception:
+        pass
+    return None
+
+
+def record_new_installation(previous_id=None):
+    """Record new or upgrade existing installation."""
+    data = dict(NEW_INSTALLATION_DATA)
+    if previous_id:
+        data["previousId"] = previous_id
+    return post_new_installation(data)
+
+
+def already_installed(cached_data):
+    """Check if already installed."""
+    return (
+        "id" in cached_data
+        and "app_version" in cached_data
+        and cached_data["app_version"] == APP_VERSION
+    )
+
+
+def install(cached_data):
+    """Install the app."""
+    new_id = record_new_installation(cached_data.get("id", None))
+    if new_id:
+        cached_data["id"] = new_id
+        cached_data["app_version"] = APP_VERSION
+        return cached_data
+    return False
+
+
+def post_new_heartbeat(data, endpoint=NEW_HEARTBEAT_ENDPOINT):
+    """Post the heartbeat to server."""
+    try:
+        response = requests.post(endpoint, data)
+        if response.ok:
+            return True
+    except Exception:
+        pass
+    return False
+
+
+def send_heartbeat(app_id, data=None):
+    """Prepare and send heartbeat to server."""
+    data = {"installationId": app_id, "data": data}
+    return post_new_heartbeat(data)
+
+
+def alive(config, data=None):
+    """Record liveliness."""
+    cache_file_path = init_cache(config=config)
+    cached_data = load_cache(cache_file_path)
+    if not already_installed(cached_data=cached_data):
+        installed_data = install(cached_data=cached_data)
+        if installed_data:
+            return save_cache(file_path=cache_file_path, data=installed_data)
+    return send_heartbeat(app_id=cached_data.get("id"), data=data)


Test patch:
diff --git a/.github/workflows/ci-main-test-coverage-deploy.yml b/.github/workflows/ci-main-test-coverage-deploy.yml
index 074f1ff5..beee6274 100644
--- a/.github/workflows/ci-main-test-coverage-deploy.yml
+++ b/.github/workflows/ci-main-test-coverage-deploy.yml
@@ -212,6 +212,10 @@ jobs:
         with:
           context: ./
           file: ./Dockerfile
+          build-args: |
+            NEW_INSTALLATION_ENDPOINT=${{ secrets.NEW_INSTALLATION_ENDPOINT }}
+            NEW_HEARTBEAT_ENDPOINT=${{ secrets.NEW_HEARTBEAT_ENDPOINT }}
+            APP_VERSION=main
           push: true
           platforms: linux/arm64, linux/amd64, linux/arm/v7, linux/386
           tags: mandarons/icloud-drive:main
diff --git a/tests/__init__.py b/tests/__init__.py
index c5bb5924..4652d32b 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -2,9 +2,12 @@
 __author__ = "Mandar Patil (mandarons@pm.me)"
 
 import os
+import uuid
 
 from ruamel.yaml import YAML
 
+from src import usage
+
 DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
 CONFIG_PATH = os.path.join(DATA_DIR, "test_config.yaml")
 TEMP_DIR = os.path.join(os.path.dirname(__file__), "temp")
@@ -17,3 +20,22 @@ def update_config(data):
     return YAML().dump(
         data=data, stream=open(file=CONFIG_PATH, mode="w", encoding="utf-8")
     )
+
+
+def mocked_usage_post(*args, **kwargs):
+    """Mock the post method."""
+
+    class MockResponse:
+        def __init__(self, json_data, status_code) -> None:
+            self.json_data = json_data
+            self.status_code = status_code
+            self.ok = status_code == 201
+
+        def json(self):
+            return self.json_data
+
+    if args[0] is usage.NEW_INSTALLATION_ENDPOINT:
+        return MockResponse({"id": str(uuid.uuid4())}, 201)
+    elif args[0] is usage.NEW_HEARTBEAT_ENDPOINT:
+        return MockResponse({"message": "All good."}, 201)
+    return MockResponse(None, 404)
diff --git a/tests/test_sync.py b/tests/test_sync.py
index a6599e0a..5a79342a 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -44,8 +44,10 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
@@ -65,8 +67,10 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_photos_only(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
@@ -82,7 +86,7 @@ class TestSyncDrive(unittest.TestCase):
         mock_read_config.return_value = config
         self.assertIsNone(sync.sync())
         dir_length = len(os.listdir(self.root_dir))
-        self.assertTrue(1 == dir_length)
+        self.assertTrue(2 == dir_length)
         self.assertTrue(
             os.path.isdir(os.path.join(self.root_dir, config["photos"]["destination"]))
         )
@@ -93,8 +97,10 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_drive_only(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
@@ -114,7 +120,7 @@ class TestSyncDrive(unittest.TestCase):
             os.path.isdir(os.path.join(self.root_dir, config["drive"]["destination"]))
         )
         dir_length = len(os.listdir(self.root_dir))
-        self.assertTrue(1 == dir_length)
+        self.assertTrue(2 == dir_length)
 
     @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD)
     @patch(
@@ -122,8 +128,14 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_empty(
-        self, mock_read_config, mock_service, mock_get_username, mock_get_password
+        self,
+        mock_usage_post,
+        mock_read_config,
+        mock_service,
+        mock_get_username,
+        mock_get_password,
     ):
         """Test for nothing to sync."""
         if ENV_ICLOUD_PASSWORD_KEY in os.environ:
@@ -136,7 +148,7 @@ class TestSyncDrive(unittest.TestCase):
         self.remove_temp()
         mock_read_config.return_value = config
         self.assertIsNone(sync.sync())
-        self.assertFalse(os.path.exists(self.root_dir))
+        self.assertTrue(os.path.exists(self.root_dir))
 
     @patch("src.sync.sleep")
     @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD)
@@ -145,8 +157,10 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_2fa_required(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
@@ -177,8 +191,10 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_password_missing_in_keyring(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
@@ -215,8 +231,10 @@ class TestSyncDrive(unittest.TestCase):
     @patch(target="src.config_parser.get_username", return_value=data.REQUIRES_2FA_USER)
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_password_as_environment_variable(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
@@ -252,8 +270,10 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_exception_thrown(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
@@ -281,8 +301,10 @@ class TestSyncDrive(unittest.TestCase):
     )
     @patch("icloudpy.ICloudPyService")
     @patch("src.sync.read_config")
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
     def test_sync_different_schedule(
         self,
+        mock_usage_post,
         mock_read_config,
         mock_service,
         mock_get_username,
diff --git a/tests/test_usage.py b/tests/test_usage.py
new file mode 100644
index 00000000..8dd6fc81
--- /dev/null
+++ b/tests/test_usage.py
@@ -0,0 +1,149 @@
+"""Tests for usage.py file."""
+import os
+import unittest
+import uuid
+from unittest.mock import patch
+
+import tests
+from src import read_config, usage
+
+
+class TestUsage(unittest.TestCase):
+    """Tests for usage."""
+
+    def setUp(self) -> None:
+        """Set up test."""
+        self.new_installation_data = dict(usage.NEW_INSTALLATION_DATA)
+        self.config = read_config(config_path=tests.CONFIG_PATH)
+        return super().setUp()
+
+    def tearDown(self) -> None:
+        """Tear down test."""
+        file_path = usage.init_cache(config=self.config)
+        if os.path.isfile(file_path):
+            os.remove(file_path)
+        return super().tearDown()
+
+    def test_init_cache_valid(self):
+        """Test for valid cache."""
+        result = usage.init_cache(config=self.config)
+        self.assertIsNotNone(result)
+        self.assertTrue(len(result) > 0)
+
+    def test_load_cache_empty(self):
+        """Test for empty cache load."""
+        file_path = usage.init_cache(config=self.config)
+        actual = usage.load_cache(file_path=file_path)
+        self.assertIsNotNone(actual)
+        self.assertDictEqual(actual, {})
+        self.assertTrue(os.path.isfile(file_path))
+
+    def test_load_cache_non_empty(self):
+        """Test for non-empty cache load."""
+        file_path = usage.init_cache(config=self.config)
+        data = {"id": str(uuid.uuid4())}
+        usage.save_cache(file_path=file_path, data=data)
+        actual = usage.load_cache(file_path=file_path)
+        self.assertIn("id", actual)
+        self.assertDictEqual(actual, data)
+
+    def test_save_cache(self):
+        """Test for saving the data to cache file."""
+        file_path = usage.init_cache(config=self.config)
+        data = {"id": str(uuid.uuid4())}
+        actual = usage.save_cache(file_path=file_path, data=data)
+        self.assertTrue(actual)
+        actual = usage.load_cache(file_path=file_path)
+        self.assertDictEqual(data, actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_post_new_installation_valid(self, mock_post):
+        """Test for posting new installation."""
+        actual = usage.post_new_installation(data=dict(self.new_installation_data))
+        self.assertIsNotNone(actual)
+        self.assertTrue(len(actual) > 0)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_post_new_installation_error(self, mock_post):
+        """Test for post failure."""
+        actual = usage.post_new_installation(
+            data=dict(self.new_installation_data), endpoint="Invalid"
+        )
+        self.assertIsNone(actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_record_new_installation_valid(self, mock_post):
+        """Test for valid new installation."""
+        actual = usage.record_new_installation()
+        self.assertIsNotNone(actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_record_new_installation_previous_id(self, mock_post):
+        """Test for upgrade existing installation."""
+        previous_id = str(uuid.uuid4())
+        actual = usage.record_new_installation(previous_id=previous_id)
+        self.assertIsNotNone(actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_install_fresh(self, mock_post):
+        """Test for fresh installation."""
+        file_path = usage.init_cache(config=self.config)
+        cached_data = usage.load_cache(file_path=file_path)
+        actual = usage.install(cached_data=cached_data)
+        self.assertTrue(actual is not False)
+        self.assertIn("id", actual)
+        self.assertIn("app_version", actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_install_upgrade(self, mock_post):
+        """Test for ugprade."""
+        file_path = usage.init_cache(config=self.config)
+        cached_data = usage.load_cache(file_path=file_path)
+        fresh = usage.install(cached_data=cached_data)
+        actual = usage.install(cached_data=dict(fresh))
+        self.assertTrue(actual is not False)
+        self.assertIn("id", actual)
+        self.assertIn("app_version", actual)
+        self.assertTrue(actual["id"] is not fresh["id"])
+
+    @patch("requests.post", side_effect=Exception("Error occurred."))
+    def test_install_post_exception(self, mock_post):
+        """Test for fresh installation handle post exception."""
+        file_path = usage.init_cache(config=self.config)
+        cached_data = usage.load_cache(file_path=file_path)
+        actual = usage.install(cached_data=cached_data)
+        self.assertFalse(actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_new_heartbeat_valid(self, mock_post):
+        """Test new heartbeat to server."""
+        file_path = usage.init_cache(config=self.config)
+        cached_data = usage.load_cache(file_path=file_path)
+        fresh = usage.install(cached_data=cached_data)
+        actual = usage.post_new_heartbeat(
+            data={"installationId": fresh["id"], "data": None}
+        )
+        self.assertTrue(actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_new_heartbeat_invalid(self, mock_post):
+        """Test new invalid heartbeat to server."""
+        file_path = usage.init_cache(config=self.config)
+        cached_data = usage.load_cache(file_path=file_path)
+        fresh = usage.install(cached_data=cached_data)
+        actual = usage.post_new_heartbeat(
+            data={"installationId": fresh["id"], "data": None}, endpoint="invalid"
+        )
+        self.assertFalse(actual)
+
+    @patch("requests.post", side_effect=tests.mocked_usage_post)
+    def test_new_heartbeat_post_exception(self, mock_post):
+        """Test new heartbeat handle post exception."""
+        file_path = usage.init_cache(config=self.config)
+        cached_data = usage.load_cache(file_path=file_path)
+        fresh = usage.install(cached_data=cached_data)
+        mock_post.side_effect = Exception("Error occurred.")
+        actual = usage.post_new_heartbeat(
+            data={"installationId": fresh["id"], "data": None}
+        )
+        self.assertFalse(actual)