Merge pull request #663 from samschott/declarative-tests

[tests] create and assert sync states declaratively

tests/linked/integration/conftest.py

@@ -5,16 +5,8 @@
 import pytest
 
 from maestral.main import Maestral
-from maestral.core import FileMetadata
 from maestral.config import remove_configuration
-from maestral.utils.path import (
-    generate_cc_name,
-    delete,
-    to_existing_unnormalized_path,
-    is_child,
-    walk,
-    get_symlink_target,
-)
+from maestral.utils.path import generate_cc_name, delete
 from maestral.utils.appdirs import get_home_dir
 from maestral.keyring import TokenType
 from maestral.exceptions import NotFoundError
@@ -63,6 +55,9 @@ def m():
     for entry in res.entries:
         m.client.remove(entry.path_lower)
 
+    # disable throttling for tests
+    m.sync.max_cpu_percent = 20 * 100
+
     # start syncing
     m.start_sync()
     wait_for_idle(m)
@@ -116,65 +111,3 @@ def wait_for_idle(m: Maestral, cycles: int = 4) -> None:
         else:
             time.sleep(1)
             count += 1
-
-
-def assert_synced(m: Maestral) -> None:
-    """Asserts that the ``local_folder`` and ``remote_folder`` are synced."""
-
-    listing = m.client.list_folder("/", recursive=True)
-
-    # Assert that all items from server are present locally with the same content hash.
-    for md in listing.entries:
-
-        if m.sync.is_excluded_by_user(md.path_lower):
-            continue
-
-        local_path = m.to_local_path(md.path_display)
-
-        remote_hash = md.content_hash if isinstance(md, FileMetadata) else "folder"
-        local_hash = m.sync.get_local_hash(local_path)
-        local_symlink_target = get_symlink_target(local_path)
-
-        assert local_hash, f"'{md.path_display}' not found locally"
-        assert local_hash == remote_hash, f'different content for "{md.path_display}"'
-
-        if isinstance(md, FileMetadata):
-            assert (
-                md.symlink_target == local_symlink_target
-            ), f'different symlink targets for "{md.path_display}"'
-
-    # Assert that all local items are present on server.
-    for path, _ in walk(m.dropbox_path, m.sync._scandir_with_ignore):
-        dbx_path = m.sync.to_dbx_path_lower(path)
-        has_match = any(md for md in listing.entries if md.path_lower == dbx_path)
-        assert has_match, f'local item "{path}" does not exist on dbx'
-
-    # Check that our index is correct.
-    for index_entry in m.sync.get_index():
-
-        if is_child(index_entry.dbx_path_lower, "/"):
-            # Check that there is a match on the server.
-            matching_items = [
-                e for e in listing.entries if e.path_lower == index_entry.dbx_path_lower
-            ]
-            assert (
-                len(matching_items) == 1
-            ), f'indexed item "{index_entry.dbx_path_lower}" does not exist on dbx'
-
-            e = matching_items[0]
-            remote_rev = e.rev if isinstance(e, FileMetadata) else "folder"
-
-            # Check if revs are equal on server and locally.
-            assert (
-                index_entry.rev == remote_rev
-            ), f'different revs for "{index_entry.dbx_path_lower}"'
-
-            # Check if casing on drive is the same as in index.
-            local_path_expected_casing = m.dropbox_path + index_entry.dbx_path_cased
-            local_path_actual_casing = to_existing_unnormalized_path(
-                local_path_expected_casing
-            )
-
-            assert (
-                local_path_expected_casing == local_path_actual_casing
-            ), "casing on drive does not match index"

tests/linked/integration/test_main.py

@@ -1,13 +1,15 @@
-import pathlib
+from __future__ import annotations
+
 import sys
 import os
 import os.path as osp
 import subprocess
 
 import pytest
+from dropbox import files
 
 from maestral.main import Maestral
-from maestral.core import FileMetadata, WriteMode
+from maestral.core import FileMetadata
 from maestral.exceptions import (
     NotFoundError,
     UnsupportedFileTypeForDiff,
@@ -18,22 +20,21 @@
 from maestral.utils.path import delete
 from maestral.utils.integration import get_inotify_limits
 
-from .conftest import wait_for_idle, resources
+from .conftest import wait_for_idle
 
 
 if not ("DROPBOX_ACCESS_TOKEN" in os.environ or "DROPBOX_REFRESH_TOKEN" in os.environ):
     pytest.skip("Requires auth token", allow_module_level=True)
 
 
 def _create_file_with_content(
-    m: Maestral, tmp_path: pathlib.Path, dbx_path: str, content: str, mode: str = "w"
+    m: Maestral, dbx_path: str, content: str | bytes
 ) -> FileMetadata:
 
-    local_path = str(tmp_path / dbx_path.lstrip("/"))
-    with open(local_path, mode) as f:
-        f.write(content)
+    if isinstance(content, str):
+        content = content.encode()
 
-    return m.client.upload(local_path, dbx_path, write_mode=WriteMode.Overwrite)
+    return m.client.dbx.files_upload(content, dbx_path, mode=files.WriteMode.overwrite)
 
 
 def test_status_properties(m: Maestral) -> None:
@@ -268,26 +269,26 @@ def test_selective_sync_nested(m: Maestral) -> None:
     assert not m.fatal_errors
 
 
-def test_get_file_diff(m: Maestral, tmp_path) -> None:
+def test_get_file_diff(m: Maestral) -> None:
     dbx_path = "/test.txt"
 
-    md_old = _create_file_with_content(m, tmp_path, dbx_path, "old")
-    md_new = _create_file_with_content(m, tmp_path, dbx_path, "new")
+    md_old = _create_file_with_content(m, dbx_path, "old")
+    md_new = _create_file_with_content(m, dbx_path, "new")
     diff = m.get_file_diff(md_old.rev, md_new.rev)
 
     assert diff[2] == "@@ -1 +1 @@\n"
     assert diff[3] == "-old"
     assert diff[4] == "+new"
 
 
-def test_get_file_diff_local(m: Maestral, tmp_path) -> None:
+def test_get_file_diff_local(m: Maestral) -> None:
     dbx_path = "/test.txt"
     local_path = m.to_local_path(dbx_path)
 
     m.stop_sync()
     wait_for_idle(m)
 
-    md_old = _create_file_with_content(m, tmp_path, dbx_path, "old")
+    md_old = _create_file_with_content(m, dbx_path, "old")
 
     with open(local_path, "w") as f:
         f.write("new")
@@ -299,46 +300,46 @@ def test_get_file_diff_local(m: Maestral, tmp_path) -> None:
     assert diff[4] == "+new"
 
 
-def test_get_file_diff_not_found(m: Maestral, tmp_path) -> None:
+def test_get_file_diff_not_found(m: Maestral) -> None:
     dbx_path = "/test.txt"
 
-    md_new = _create_file_with_content(m, tmp_path, dbx_path, "new")
+    md_new = _create_file_with_content(m, dbx_path, "new")
 
     with pytest.raises(NotFoundError):
         m.get_file_diff("015db1e6dec9da000000001f7709020", md_new.rev)
 
 
-def test_get_file_diff_unsupported_ext(m: Maestral, tmp_path) -> None:
+def test_get_file_diff_unsupported_ext(m: Maestral) -> None:
     """Tests file diffs for unsupported file types."""
 
     dbx_path = "/test.pdf"
-    md_old = _create_file_with_content(m, tmp_path, dbx_path, "old")
-    md_new = _create_file_with_content(m, tmp_path, dbx_path, "new")
+    md_old = _create_file_with_content(m, dbx_path, "old")
+    md_new = _create_file_with_content(m, dbx_path, "new")
 
     with pytest.raises(UnsupportedFileTypeForDiff):
         m.get_file_diff(md_old.rev, md_new.rev)
 
 
-def test_get_file_diff_unsupported_content(m: Maestral, tmp_path) -> None:
+def test_get_file_diff_unsupported_content(m: Maestral) -> None:
     """Tests file diffs for unsupported file types."""
 
     dbx_path = "/test.txt"
     # Upload a compiled c file with .txt extension
-    md_old = m.client.upload(resources + "/bin.txt", dbx_path)
-    md_new = _create_file_with_content(m, tmp_path, dbx_path, "new")
+    md_old = _create_file_with_content(m, dbx_path, b"\xcf\xfa\xed\xfe\x07")
+    md_new = _create_file_with_content(m, dbx_path, "new")
 
     with pytest.raises(UnsupportedFileTypeForDiff):
         m.get_file_diff(md_old.rev, md_new.rev)
 
 
-def test_get_file_diff_unsupported_content_local(m: Maestral, tmp_path) -> None:
+def test_get_file_diff_unsupported_content_local(m: Maestral) -> None:
     dbx_path = "/test.txt"
     local_path = m.to_local_path(dbx_path)
 
     m.stop_sync()
     wait_for_idle(m)
 
-    md_old = _create_file_with_content(m, tmp_path, dbx_path, "old")
+    md_old = _create_file_with_content(m, dbx_path, "old")
 
     with open(local_path, "wb") as f:
         f.write("möglich".encode("cp273"))