Check storage availability only for a project owned by a current user (#70)

Ownership depends on matching username with namespace, not ownership permissions

Author: sklencar

.github/workflows/autotests.yml

@@ -4,6 +4,8 @@ env:
   TEST_MERGIN_URL: https://test.dev.cloudmergin.com/
   TEST_API_USERNAME: test_plugin
   TEST_API_PASSWORD: ${{ secrets.MERGINTEST_API_PASSWORD }}
+  TEST_API_USERNAME2: test_plugin2
+  TEST_API_PASSWORD2: ${{ secrets.MERGINTEST_API_PASSWORD2 }}
 
 jobs:
   tests:

README.md

@@ -118,4 +118,6 @@ For running test do:
     export TEST_MERGIN_URL=<url> # testing server
     export TEST_API_USERNAME=<username>
     export TEST_API_PASSWORD=<pwd>
+    export TEST_API_USERNAME2=<username2>
+    export TEST_API_PASSWORD2=<pwd2>
     pipenv run pytest --cov-report html --cov=mergin test/

mergin/client.py

@@ -373,6 +373,27 @@ def user_info(self):
         resp = self.get('/v1/user/' + self.username())
         return json.load(resp)
 
+    def set_project_access(self, project_path, access):
+        """
+        Updates access for given project.
+        :param project_path: project full name (<namespace>/<name>)
+        :param access: dict <readersnames, writersnames, ownersnames> -> list of str username we want to give access to
+        """
+        if not self._user_info:
+            raise Exception("Authentication required")
+
+        params = {"access": access}
+        path = "/v1/project/%s" % project_path
+        url = urllib.parse.urljoin(self.url, urllib.parse.quote(path))
+        json_headers = {'Content-Type': 'application/json'}
+        try:
+            request = urllib.request.Request(url, data=json.dumps(params).encode(), headers=json_headers, method="PUT")
+            self._do_request(request)
+        except Exception as e:
+            detail = f"Project path: {project_path}"
+            raise ClientError(str(e), detail)
+
+
     def push_project(self, directory):
         """
         Upload local changes to the repository.

mergin/client_push.py

@@ -101,11 +101,14 @@ def push_project_async(mc, directory):
 
     changes = mp.get_push_changes()
     mp.log.debug("push changes:\n" + pprint.pformat(changes))
-    enough_free_space, freespace = mc.enough_storage_available(changes)
-    if not enough_free_space:
-        freespace = int(freespace/(1024*1024))
-        mp.log.error(f"--- push {project_path} - not enough space")
-        raise ClientError("Storage limit has been reached. Only " + str(freespace) + "MB left")
+
+    # currently proceed storage limit check only if a project is own by a current user.
+    if username == project_path.split("/")[0]:
+        enough_free_space, freespace = mc.enough_storage_available(changes)
+        if not enough_free_space:
+            freespace = int(freespace/(1024*1024))
+            mp.log.error(f"--- push {project_path} - not enough space")
+            raise ClientError("Storage limit has been reached. Only " + str(freespace) + "MB left")
 
     if not sum(len(v) for v in changes.values()):
         mp.log.info(f"--- push {project_path} - nothing to do")

mergin/test/test_client.py

@@ -11,6 +11,8 @@
 SERVER_URL = os.environ.get('TEST_MERGIN_URL')
 API_USER = os.environ.get('TEST_API_USERNAME')
 USER_PWD = os.environ.get('TEST_API_PASSWORD')
+API_USER2 = os.environ.get('TEST_API_USERNAME2')
+USER_PWD2 = os.environ.get('TEST_API_PASSWORD2')
 TMP_DIR = tempfile.gettempdir()
 TEST_DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'test_data')
 CHANGED_SCHEMA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'modified_schema')
@@ -22,8 +24,16 @@ def toggle_geodiff(enabled):
 
 @pytest.fixture(scope='function')
 def mc():
-    assert SERVER_URL and SERVER_URL.rstrip('/') != 'https://public.cloudmergin.com' and API_USER and USER_PWD
-    return MerginClient(SERVER_URL, login=API_USER, password=USER_PWD)
+    return create_client(API_USER, USER_PWD)
+
+@pytest.fixture(scope='function')
+def mc2():
+    return create_client(API_USER2, USER_PWD2)
+
+
+def create_client(user, pwd):
+    assert SERVER_URL and SERVER_URL.rstrip('/') != 'https://public.cloudmergin.com' and user and pwd
+    return MerginClient(SERVER_URL, login=user, password=pwd)
 
 
 def cleanup(mc, project, dirs):
@@ -32,6 +42,11 @@ def cleanup(mc, project, dirs):
         mc.delete_project(project)
     except ClientError:
         pass
+    remove_folders(dirs)
+
+
+def remove_folders(dirs):
+    # clean given directories
     for d in dirs:
         if os.path.exists(d):
             shutil.rmtree(d)
@@ -461,6 +476,7 @@ def test_empty_file_in_subdir(mc):
     mc.pull_project(project_dir_2)
     assert os.path.exists(os.path.join(project_dir_2, 'subdir2', 'empty2.txt'))
 
+
 def test_clone_project(mc):
     test_project = 'test_clone_project'
     test_project_fullname = API_USER + '/' + test_project
@@ -483,3 +499,148 @@ def test_clone_project(mc):
     mc.clone_project(test_project_fullname, cloned_project_name, API_USER)
     projects = mc.projects_list(flag='created')
     assert any(p for p in projects if p['name'] == cloned_project_name and p['namespace'] == API_USER)
+
+
+def test_set_read_write_access(mc):
+    test_project = 'test_set_read_write_access'
+    test_project_fullname = API_USER + '/' + test_project
+
+    # cleanups
+    project_dir = os.path.join(TMP_DIR, test_project, API_USER)
+    cleanup(mc, test_project_fullname, [project_dir])
+
+    # create new (empty) project on server
+    mc.create_project(test_project)
+
+    # Add writer access to another client
+    project_info = get_project_info(mc, API_USER, test_project)
+    access = project_info['access']
+    access['writersnames'].append(API_USER2)
+    access['readersnames'].append(API_USER2)
+    mc.set_project_access(test_project_fullname, access)
+
+    # check access
+    project_info = get_project_info(mc, API_USER, test_project)
+    access = project_info['access']
+    assert API_USER2 in access['writersnames']
+    assert API_USER2 in access['readersnames']
+
+
+def test_available_storage_validation(mc):
+    """
+    Testing of storage limit - applies to user pushing changes into own project (namespace matching username).
+    This test also tests giving read and write access to another user. Additionally tests also uploading of big file.
+    """
+    test_project = 'test_available_storage_validation'
+    test_project_fullname = API_USER + '/' + test_project
+
+    # cleanups
+    project_dir = os.path.join(TMP_DIR, test_project, API_USER)
+    cleanup(mc, test_project_fullname, [project_dir])
+
+    # create new (empty) project on server
+    mc.create_project(test_project)
+
+    # download project
+    mc.download_project(test_project_fullname, project_dir)
+
+    # get user_info about storage capacity
+    user_info = mc.user_info()
+    storage_remaining = user_info['storage'] - user_info['disk_usage']
+
+    # generate dummy data (remaining storage + extra 1024b)
+    dummy_data_path = project_dir + "/data"
+    file_size = storage_remaining + 1024
+    _generate_big_file(dummy_data_path, file_size)
+
+    # try to upload
+    got_right_err = False
+    try:
+        mc.push_project(project_dir)
+    except ClientError as e:
+        # Expecting "Storage limit has been reached" error msg.
+        assert str(e).startswith("Storage limit has been reached")
+        got_right_err = True    
+    assert got_right_err
+
+    # Expecting empty project
+    project_info = get_project_info(mc, API_USER, test_project)
+    assert project_info['meta']['files_count'] == 0
+    assert project_info['meta']['size'] == 0
+
+
+def test_available_storage_validation2(mc, mc2):
+    """
+    Testing of storage limit - should not be applied for user pushing changes into project with different namespace.
+    This should cover the exception of mergin-py-client that a user can push changes to someone else's project regardless
+    the user's own storage limitation. Of course, other limitations are still applied (write access, owner of
+    a modified project has to have enough free storage).
+
+    Therefore NOTE that there are following assumptions:
+        - API_USER2's free storage >= API_USER's free storage + 1024b (size of changes to be pushed)
+        - both accounts should ideally have a free plan
+    """
+    test_project = 'test_available_storage_validation2'
+    test_project_fullname = API_USER2 + '/' + test_project
+
+    # cleanups
+    project_dir = os.path.join(TMP_DIR, test_project, API_USER)
+    cleanup(mc, test_project_fullname, [project_dir])
+    cleanup(mc2, test_project_fullname, [project_dir])
+
+    # create new (empty) project on server
+    mc2.create_project(test_project)
+
+    # Add writer access to another client
+    project_info = get_project_info(mc2, API_USER2, test_project)
+    access = project_info['access']
+    access['writersnames'].append(API_USER)
+    access['readersnames'].append(API_USER)
+    mc2.set_project_access(test_project_fullname, access)
+
+    # download project
+    mc.download_project(test_project_fullname, project_dir)
+
+    # get user_info about storage capacity
+    user_info = mc.user_info()
+    storage_remaining = user_info['storage'] - user_info['disk_usage']
+
+    # generate dummy data (remaining storage + extra 1024b)
+    dummy_data_path = project_dir + "/data"
+    file_size = storage_remaining + 1024
+    _generate_big_file(dummy_data_path, file_size)
+
+    # try to upload
+    mc.push_project(project_dir)
+
+    # Check project content
+    project_info = get_project_info(mc2, API_USER2, test_project)
+    assert project_info['meta']['files_count'] == 1
+    assert project_info['meta']['size'] == file_size
+
+    # remove dummy big file from a disk
+    remove_folders([project_dir])
+
+
+def get_project_info(mc, namespace, project_name):
+    """
+    Returns first (and suppose to be just one) project info dict of project matching given namespace and name.
+    :param mc: MerginClient instance
+    :param namespace: project's namespace
+    :param project_name: project's name
+    :return: dict with project info
+    """
+    projects = mc.projects_list(flag='created')
+    test_project_list = [p for p in projects if p['name'] == project_name and p['namespace'] == namespace]
+    assert len(test_project_list) == 1
+    return test_project_list[0]
+
+
+def _generate_big_file(filepath, size):
+    """
+    generate big binary file with the specified size in bytes
+    :param filepath: full filepath
+    :param size: the size in bytes
+    """
+    with open(filepath, 'wb') as fout:
+        fout.write(b"\0" * size)