Merge pull request #23856 from Connor9220/library-editor
CAM: Replace the main library editor dialog and add copy & paste & drag & drop support
This commit is contained in:
@@ -25,6 +25,20 @@ import Path.Base.PropertyBag as PathPropertyBag
|
||||
import CAMTests.PathTestUtils as PathTestUtils
|
||||
|
||||
|
||||
def as_group_list(groups):
|
||||
"""Normalize CustomPropertyGroups to a list of strings."""
|
||||
if groups is None:
|
||||
return []
|
||||
if isinstance(groups, (list, tuple)):
|
||||
return list(groups)
|
||||
if isinstance(groups, str):
|
||||
return [groups]
|
||||
try:
|
||||
return list(groups)
|
||||
except Exception:
|
||||
return [str(groups)]
|
||||
|
||||
|
||||
class TestPathPropertyBag(PathTestUtils.PathTestBase):
|
||||
def setUp(self):
|
||||
self.doc = FreeCAD.newDocument("test-property-bag")
|
||||
@@ -37,7 +51,7 @@ class TestPathPropertyBag(PathTestUtils.PathTestBase):
|
||||
bag = PathPropertyBag.Create()
|
||||
self.assertTrue(hasattr(bag, "Proxy"))
|
||||
self.assertEqual(bag.Proxy.getCustomProperties(), [])
|
||||
self.assertEqual(bag.CustomPropertyGroups, [])
|
||||
self.assertEqual(as_group_list(bag.CustomPropertyGroups), [])
|
||||
|
||||
def test01(self):
|
||||
"""adding properties to a PropertyBag is tracked properly"""
|
||||
@@ -48,7 +62,7 @@ class TestPathPropertyBag(PathTestUtils.PathTestBase):
|
||||
bag.Title = "Madame"
|
||||
self.assertEqual(bag.Title, "Madame")
|
||||
self.assertEqual(bag.Proxy.getCustomProperties(), ["Title"])
|
||||
self.assertEqual(bag.CustomPropertyGroups, ["Address"])
|
||||
self.assertEqual(as_group_list(bag.CustomPropertyGroups), ["Address"])
|
||||
|
||||
def test02(self):
|
||||
"""refreshCustomPropertyGroups deletes empty groups"""
|
||||
@@ -59,7 +73,7 @@ class TestPathPropertyBag(PathTestUtils.PathTestBase):
|
||||
bag.removeProperty("Title")
|
||||
proxy.refreshCustomPropertyGroups()
|
||||
self.assertEqual(bag.Proxy.getCustomProperties(), [])
|
||||
self.assertEqual(bag.CustomPropertyGroups, [])
|
||||
self.assertEqual(as_group_list(bag.CustomPropertyGroups), [])
|
||||
|
||||
def test03(self):
|
||||
"""refreshCustomPropertyGroups does not delete non-empty groups"""
|
||||
@@ -72,4 +86,4 @@ class TestPathPropertyBag(PathTestUtils.PathTestBase):
|
||||
bag.removeProperty("Gender")
|
||||
proxy.refreshCustomPropertyGroups()
|
||||
self.assertEqual(bag.Proxy.getCustomProperties(), ["Title"])
|
||||
self.assertEqual(bag.CustomPropertyGroups, ["Address"])
|
||||
self.assertEqual(as_group_list(bag.CustomPropertyGroups), ["Address"])
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
"""
|
||||
AssetManager tests.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
import asyncio
|
||||
from unittest.mock import Mock
|
||||
import pathlib
|
||||
import tempfile
|
||||
from typing import Any, Mapping, List
|
||||
from typing import Any, Mapping, List, Type, Optional, cast
|
||||
from Path.Tool.assets import (
|
||||
AssetManager,
|
||||
FileStore,
|
||||
@@ -24,7 +28,7 @@ class MockAsset(Asset):
|
||||
self._id = id
|
||||
|
||||
@classmethod
|
||||
def extract_dependencies(cls, data: bytes, serializer: AssetSerializer) -> List[AssetUri]:
|
||||
def extract_dependencies(cls, data: bytes, serializer: Type[AssetSerializer]) -> List[AssetUri]:
|
||||
# Mock implementation doesn't use data or format for dependencies
|
||||
return []
|
||||
|
||||
@@ -33,18 +37,83 @@ class MockAsset(Asset):
|
||||
cls,
|
||||
data: bytes,
|
||||
id: str,
|
||||
dependencies: Mapping[AssetUri, Asset] | None,
|
||||
serializer: AssetSerializer,
|
||||
dependencies: Optional[Mapping[AssetUri, Asset]],
|
||||
serializer: Type[AssetSerializer],
|
||||
) -> "MockAsset":
|
||||
# Create instance with provided id
|
||||
return cls(data, id)
|
||||
|
||||
def to_bytes(self, serializer: AssetSerializer) -> bytes:
|
||||
def to_bytes(self, serializer: Type[AssetSerializer]) -> bytes:
|
||||
return self._data
|
||||
|
||||
def get_id(self) -> str:
|
||||
return self._id
|
||||
|
||||
def get_data(self) -> bytes:
|
||||
"""Returns the raw data stored in the mock asset."""
|
||||
return self._data
|
||||
|
||||
|
||||
# Mock Asset class with dependencies for testing deepcopy
|
||||
class MockAssetWithDeps(Asset):
|
||||
asset_type: str = "mock_asset_with_deps"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
data: Any = None,
|
||||
id: str = "mock_id",
|
||||
dependencies: Optional[Mapping[AssetUri, Asset]] = None,
|
||||
):
|
||||
self._data = data
|
||||
self._id = id
|
||||
self._dependencies = dependencies or {}
|
||||
|
||||
@classmethod
|
||||
def extract_dependencies(cls, data: bytes, serializer: Type[AssetSerializer]) -> List[AssetUri]:
|
||||
# Assuming data is a simple JSON string like '{"deps": ["uri1", "uri2"]}'
|
||||
try:
|
||||
import json
|
||||
|
||||
data_str = data.decode("utf-8")
|
||||
data_dict = json.loads(data_str)
|
||||
dep_uris_str = data_dict.get("deps", [])
|
||||
return [AssetUri(uri_str) for uri_str in dep_uris_str]
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
@classmethod
|
||||
def from_bytes(
|
||||
cls,
|
||||
data: bytes,
|
||||
id: str,
|
||||
dependencies: Optional[Mapping[AssetUri, Asset]],
|
||||
serializer: Type[AssetSerializer],
|
||||
) -> "MockAssetWithDeps":
|
||||
# Create instance with provided id and resolved dependencies
|
||||
return cls(data, id, dependencies)
|
||||
|
||||
def to_bytes(self, serializer: Type[AssetSerializer]) -> bytes:
|
||||
# Serialize data and dependency URIs into a simple format
|
||||
try:
|
||||
import json
|
||||
|
||||
dep_uri_strs = [str(uri) for uri in self._dependencies.keys()]
|
||||
data_dict = {"data": self._data.decode("utf-8"), "deps": dep_uri_strs}
|
||||
return json.dumps(data_dict).encode("utf-8")
|
||||
except Exception:
|
||||
return self._data # Fallback if serialization fails
|
||||
|
||||
def get_id(self) -> str:
|
||||
return self._id
|
||||
|
||||
def get_data(self) -> bytes:
|
||||
"""Returns the raw data stored in the mock asset."""
|
||||
return self._data
|
||||
|
||||
def get_dependencies(self) -> Mapping[AssetUri, Asset]:
|
||||
"""Returns the resolved dependencies."""
|
||||
return self._dependencies
|
||||
|
||||
|
||||
class TestPathToolAssetManager(unittest.TestCase):
|
||||
def test_register_store(self):
|
||||
@@ -85,12 +154,12 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
cls,
|
||||
data: bytes,
|
||||
id: str,
|
||||
dependencies: Mapping[AssetUri, Asset] | None,
|
||||
serializer: AssetSerializer,
|
||||
dependencies: Optional[Mapping[AssetUri, Asset]],
|
||||
serializer: Type[AssetSerializer],
|
||||
) -> "AnotherMockAsset":
|
||||
return cls()
|
||||
|
||||
def to_bytes(self, serializer: AssetSerializer) -> bytes:
|
||||
def to_bytes(self, serializer: Type[AssetSerializer]) -> bytes:
|
||||
return b""
|
||||
|
||||
def get_id(self) -> str:
|
||||
@@ -134,12 +203,11 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
)
|
||||
|
||||
# Call AssetManager.get
|
||||
retrieved_object = manager.get(test_uri)
|
||||
retrieved_object = cast(MockAsset, manager.get(test_uri))
|
||||
|
||||
# Assert the retrieved object is an instance of MockAsset
|
||||
self.assertIsInstance(retrieved_object, MockAsset)
|
||||
# Assert the data was passed to from_bytes
|
||||
self.assertEqual(retrieved_object._data, test_data)
|
||||
self.assertEqual(retrieved_object.get_data(), test_data)
|
||||
|
||||
# Test error handling for non-existent URI
|
||||
non_existent_uri = AssetUri.build(MockAsset.asset_type, "non_existent", "1")
|
||||
@@ -210,7 +278,7 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
|
||||
# Verify the asset was created
|
||||
retrieved_data = asyncio.run(local_store.get(created_uri))
|
||||
self.assertEqual(retrieved_data, test_obj.to_bytes(DummyAssetSerializer))
|
||||
self.assertEqual(retrieved_data, test_obj.get_data())
|
||||
|
||||
# Test error handling (store not found)
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
@@ -244,9 +312,10 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
self.assertEqual(updated_uri.version, "2")
|
||||
|
||||
# Verify the asset was updated
|
||||
obj = manager.get(updated_uri, store="local")
|
||||
self.assertEqual(updated_data, test_obj.to_bytes(DummyAssetSerializer))
|
||||
self.assertEqual(updated_data, obj.to_bytes(DummyAssetSerializer))
|
||||
obj = cast(MockAsset, manager.get(updated_uri, store="local"))
|
||||
self.assertEqual(updated_data, test_obj.get_data())
|
||||
self.assertIsInstance(obj, MockAsset)
|
||||
self.assertEqual(updated_data, obj.get_data())
|
||||
|
||||
# Test error handling (store not found)
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
@@ -382,23 +451,17 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
uris = [uri1, uri2, uri3]
|
||||
|
||||
# Call manager.get_bulk
|
||||
retrieved_assets = manager.get_bulk(uris, store="memory_bulk")
|
||||
retrieved_assets = cast(List[MockAsset], manager.get_bulk(uris, store="memory_bulk"))
|
||||
|
||||
# Assert the correct number of assets were returned
|
||||
self.assertEqual(len(retrieved_assets), 3)
|
||||
|
||||
# Assert the retrieved assets are MockAsset instances with correct data
|
||||
self.assertIsInstance(retrieved_assets[0], MockAsset)
|
||||
self.assertEqual(
|
||||
retrieved_assets[0].to_bytes(DummyAssetSerializer),
|
||||
data1,
|
||||
)
|
||||
self.assertEqual(retrieved_assets[0].get_data(), data1)
|
||||
|
||||
self.assertIsInstance(retrieved_assets[1], MockAsset)
|
||||
self.assertEqual(
|
||||
retrieved_assets[1].to_bytes(DummyAssetSerializer),
|
||||
data2,
|
||||
)
|
||||
self.assertEqual(retrieved_assets[1].get_data(), data2)
|
||||
|
||||
# Assert the non-existent asset is None
|
||||
self.assertIsNone(retrieved_assets[2])
|
||||
@@ -442,8 +505,9 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
manager_filtered.add_raw(MockAsset.asset_type, "id2", data2, "memory_fetch_filtered")
|
||||
manager_filtered.add_raw("another_type", "id3", b"data for id3", "memory_fetch_filtered")
|
||||
|
||||
retrieved_assets_filtered = manager_filtered.fetch(
|
||||
asset_type=MockAsset.asset_type, store="memory_fetch_filtered"
|
||||
retrieved_assets_filtered = cast(
|
||||
List[MockAsset],
|
||||
manager_filtered.fetch(asset_type=MockAsset.asset_type, store="memory_fetch_filtered"),
|
||||
)
|
||||
|
||||
# Assert the correct number of assets were returned
|
||||
@@ -452,13 +516,13 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
# Assert the retrieved assets are MockAsset instances with correct data
|
||||
self.assertIsInstance(retrieved_assets_filtered[0], MockAsset)
|
||||
self.assertEqual(
|
||||
retrieved_assets_filtered[0].to_bytes(DummyAssetSerializer).decode("utf-8"),
|
||||
retrieved_assets_filtered[0].get_data().decode("utf-8"),
|
||||
data1.decode("utf-8"),
|
||||
)
|
||||
|
||||
self.assertIsInstance(retrieved_assets_filtered[1], MockAsset)
|
||||
self.assertEqual(
|
||||
retrieved_assets_filtered[1].to_bytes(DummyAssetSerializer).decode("utf-8"),
|
||||
retrieved_assets_filtered[1].get_data().decode("utf-8"),
|
||||
data2.decode("utf-8"),
|
||||
)
|
||||
|
||||
@@ -467,6 +531,340 @@ class TestPathToolAssetManager(unittest.TestCase):
|
||||
manager.fetch(store="non_existent_store")
|
||||
self.assertIn("No store registered for name:", str(cm.exception))
|
||||
|
||||
def test_copy(self):
|
||||
# Setup AssetManager with two MemoryStores and MockAsset class
|
||||
memory_store_src = MemoryStore("memory_copy_src")
|
||||
memory_store_dest = MemoryStore("memory_copy_dest")
|
||||
manager = AssetManager()
|
||||
manager.register_store(memory_store_src)
|
||||
manager.register_store(memory_store_dest)
|
||||
manager.register_asset(MockAsset, DummyAssetSerializer)
|
||||
|
||||
# Create a source asset
|
||||
src_data = b"source asset data"
|
||||
src_uri = manager.add_raw(MockAsset.asset_type, "source_id", src_data, "memory_copy_src")
|
||||
|
||||
# Test copying to a different store with default destination URI
|
||||
copied_uri_default_dest = manager.copy(
|
||||
src_uri, dest_store="memory_copy_dest", store="memory_copy_src"
|
||||
)
|
||||
self.assertEqual(copied_uri_default_dest.asset_type, src_uri.asset_type)
|
||||
self.assertEqual(copied_uri_default_dest.asset_id, src_uri.asset_id)
|
||||
self.assertEqual(copied_uri_default_dest.version, "1") # First version in dest
|
||||
|
||||
# Verify the copied asset exists in the destination store
|
||||
copied_data_default_dest = manager.get_raw(
|
||||
copied_uri_default_dest, store="memory_copy_dest"
|
||||
)
|
||||
self.assertEqual(copied_data_default_dest, src_data)
|
||||
|
||||
# Test copying to a different store with a specified destination URI
|
||||
dest_uri_specified = AssetUri.build(MockAsset.asset_type, "specified_dest_id", "1")
|
||||
copied_uri_specified_dest = manager.copy(
|
||||
src_uri,
|
||||
dest_store="memory_copy_dest",
|
||||
store="memory_copy_src",
|
||||
dest=dest_uri_specified,
|
||||
)
|
||||
self.assertEqual(copied_uri_specified_dest, dest_uri_specified)
|
||||
|
||||
# Verify the copied asset exists at the specified destination URI
|
||||
copied_data_specified_dest = manager.get_raw(
|
||||
copied_uri_specified_dest, store="memory_copy_dest"
|
||||
)
|
||||
self.assertEqual(copied_data_specified_dest, src_data)
|
||||
|
||||
# Test copying to the same store with a different destination URI
|
||||
dest_uri_same_store = AssetUri.build(MockAsset.asset_type, "same_store_dest", "1")
|
||||
copied_uri_same_store = manager.copy(
|
||||
src_uri, dest_store="memory_copy_src", store="memory_copy_src", dest=dest_uri_same_store
|
||||
)
|
||||
self.assertEqual(copied_uri_same_store, dest_uri_same_store)
|
||||
|
||||
# Verify the copied asset exists in the same store at the new URI
|
||||
copied_data_same_store = manager.get_raw(copied_uri_same_store, store="memory_copy_src")
|
||||
self.assertEqual(copied_data_same_store, src_data)
|
||||
|
||||
# Test assertion for source and destination being the same
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
manager.copy(
|
||||
src_uri, dest_store="memory_copy_src", store="memory_copy_src", dest=src_uri
|
||||
)
|
||||
self.assertIn(
|
||||
"Source and destination cannot be the same asset in the same store.",
|
||||
str(cm.exception),
|
||||
)
|
||||
|
||||
# Test overwriting existing destination (add a different asset at specified_dest_id)
|
||||
overwrite_data = b"data to be overwritten"
|
||||
overwrite_uri = manager.add_raw(
|
||||
MockAsset.asset_type, "specified_dest_id", overwrite_data, "memory_copy_dest"
|
||||
)
|
||||
self.assertEqual(overwrite_uri.version, "2") # Should be version 2 now
|
||||
|
||||
# Copy the original src_uri to the existing destination
|
||||
copied_uri_overwrite = manager.copy(
|
||||
src_uri,
|
||||
dest_store="memory_copy_dest",
|
||||
store="memory_copy_src",
|
||||
dest=dest_uri_specified,
|
||||
)
|
||||
# The version should be incremented again due to overwrite
|
||||
self.assertEqual(copied_uri_overwrite.version, "3")
|
||||
|
||||
# Verify the destination now contains the source data
|
||||
retrieved_overwritten_data = manager.get_raw(copied_uri_overwrite, store="memory_copy_dest")
|
||||
self.assertEqual(retrieved_overwritten_data, src_data)
|
||||
|
||||
def test_deepcopy(self):
|
||||
# Setup AssetManager with two MemoryStores and MockAssetWithDeps class
|
||||
memory_store_src = MemoryStore("memory_deepcopy_src")
|
||||
memory_store_dest = MemoryStore("memory_deepcopy_dest")
|
||||
manager = AssetManager()
|
||||
manager.register_store(memory_store_src)
|
||||
manager.register_store(memory_store_dest)
|
||||
manager.register_asset(MockAssetWithDeps, DummyAssetSerializer)
|
||||
|
||||
# Create dependency assets in the source store
|
||||
dep1_data = b'{"data": "dependency 1 data", "deps": []}'
|
||||
dep2_data = b'{"data": "dependency 2 data", "deps": []}'
|
||||
dep1_uri = manager.add_raw(
|
||||
MockAssetWithDeps.asset_type, "dep1_id", dep1_data, "memory_deepcopy_src"
|
||||
)
|
||||
dep2_uri = manager.add_raw(
|
||||
MockAssetWithDeps.asset_type, "dep2_id", dep2_data, "memory_deepcopy_src"
|
||||
)
|
||||
|
||||
# Create a source asset with dependencies
|
||||
src_data = (
|
||||
b'{"data": "source asset data", "deps": ["'
|
||||
+ str(dep1_uri).encode("utf-8")
|
||||
+ b'", "'
|
||||
+ str(dep2_uri).encode("utf-8")
|
||||
+ b'"]}'
|
||||
)
|
||||
src_uri = manager.add_raw(
|
||||
MockAssetWithDeps.asset_type, "source_id", src_data, "memory_deepcopy_src"
|
||||
)
|
||||
|
||||
# Test deep copying to a different store with default destination URI
|
||||
copied_uri_default_dest = manager.deepcopy(
|
||||
src_uri, dest_store="memory_deepcopy_dest", store="memory_deepcopy_src"
|
||||
)
|
||||
self.assertEqual(copied_uri_default_dest.asset_type, src_uri.asset_type)
|
||||
self.assertEqual(copied_uri_default_dest.asset_id, src_uri.asset_id)
|
||||
self.assertEqual(copied_uri_default_dest.version, "1") # First version in dest
|
||||
|
||||
# Verify the copied top-level asset exists in the destination store
|
||||
copied_asset_default_dest = cast(
|
||||
MockAssetWithDeps, manager.get(copied_uri_default_dest, store="memory_deepcopy_dest")
|
||||
)
|
||||
self.assertIsInstance(copied_asset_default_dest, MockAssetWithDeps)
|
||||
# The copied asset's data should be the serialized form including dependencies
|
||||
expected_data = b'{"data": "source asset data", "deps": ["mock_asset_with_deps://dep1_id/1", "mock_asset_with_deps://dep2_id/1"]}'
|
||||
self.assertEqual(copied_asset_default_dest.get_data(), expected_data)
|
||||
|
||||
# Verify dependencies were also copied and resolved correctly
|
||||
copied_deps_default_dest = copied_asset_default_dest.get_dependencies()
|
||||
self.assertEqual(len(copied_deps_default_dest), 2)
|
||||
self.assertIn(dep1_uri, copied_deps_default_dest)
|
||||
self.assertIn(dep2_uri, copied_deps_default_dest)
|
||||
|
||||
copied_dep1 = cast(MockAssetWithDeps, copied_deps_default_dest[dep1_uri])
|
||||
self.assertIsInstance(copied_dep1, MockAssetWithDeps)
|
||||
self.assertEqual(copied_dep1.get_data(), b'{"data": "dependency 1 data", "deps": []}')
|
||||
|
||||
copied_dep2 = cast(MockAssetWithDeps, copied_deps_default_dest[dep2_uri])
|
||||
self.assertIsInstance(copied_dep2, MockAssetWithDeps)
|
||||
self.assertEqual(copied_dep2.get_data(), b'{"data": "dependency 2 data", "deps": []}')
|
||||
|
||||
# Test deep copying with a specified destination URI for the top-level asset
|
||||
dest_uri_specified = AssetUri.build(MockAssetWithDeps.asset_type, "specified_dest_id", "1")
|
||||
copied_uri_specified_dest = manager.deepcopy(
|
||||
src_uri,
|
||||
dest_store="memory_deepcopy_dest",
|
||||
store="memory_deepcopy_src",
|
||||
dest=dest_uri_specified,
|
||||
)
|
||||
self.assertEqual(copied_uri_specified_dest, dest_uri_specified)
|
||||
|
||||
# Verify the copied asset exists at the specified destination URI
|
||||
copied_asset_specified_dest = cast(
|
||||
MockAssetWithDeps, manager.get(copied_uri_specified_dest, store="memory_deepcopy_dest")
|
||||
)
|
||||
self.assertIsInstance(copied_asset_specified_dest, MockAssetWithDeps)
|
||||
self.assertEqual(
|
||||
copied_asset_specified_dest.get_data(),
|
||||
b'{"data": "source asset data", "deps": ["mock_asset_with_deps://dep1_id/1", "mock_asset_with_deps://dep2_id/1"]}',
|
||||
)
|
||||
|
||||
# Verify dependencies were copied and resolved correctly (their URIs should be
|
||||
# in the destination store, but their asset_type and asset_id should be the same)
|
||||
copied_deps_specified_dest = copied_asset_specified_dest.get_dependencies()
|
||||
self.assertEqual(len(copied_deps_specified_dest), 2)
|
||||
|
||||
# The keys in the dependencies mapping should be the *original* URIs,
|
||||
# but the values should be the *copied* dependency assets.
|
||||
self.assertIn(dep1_uri, copied_deps_specified_dest)
|
||||
self.assertIn(dep2_uri, copied_deps_specified_dest)
|
||||
|
||||
copied_dep1_specified = cast(MockAssetWithDeps, copied_deps_specified_dest[dep1_uri])
|
||||
self.assertIsInstance(copied_dep1_specified, MockAssetWithDeps)
|
||||
self.assertEqual(
|
||||
copied_dep1_specified.get_data(), b'{"data": "dependency 1 data", "deps": []}'
|
||||
)
|
||||
# Check the URI of the copied dependency in the destination store
|
||||
self.assertIsNotNone(
|
||||
manager.get_or_none(copied_dep1_specified.get_uri(), store="memory_deepcopy_dest")
|
||||
)
|
||||
self.assertEqual(copied_dep1_specified.get_uri().asset_type, dep1_uri.asset_type)
|
||||
self.assertEqual(copied_dep1_specified.get_uri().asset_id, dep1_uri.asset_id)
|
||||
|
||||
copied_dep2_specified = cast(MockAssetWithDeps, copied_deps_specified_dest[dep2_uri])
|
||||
self.assertIsInstance(copied_dep2_specified, MockAssetWithDeps)
|
||||
self.assertEqual(
|
||||
copied_dep2_specified.get_data(), b'{"data": "dependency 2 data", "deps": []}'
|
||||
)
|
||||
# Check the URI of the copied dependency in the destination store
|
||||
self.assertIsNotNone(
|
||||
manager.get_or_none(copied_dep2_specified.get_uri(), store="memory_deepcopy_dest")
|
||||
)
|
||||
self.assertEqual(copied_dep2_specified.get_uri().asset_type, dep2_uri.asset_type)
|
||||
self.assertEqual(copied_dep2_specified.get_uri().asset_id, dep2_uri.asset_id)
|
||||
|
||||
# Test handling of existing dependencies in the destination store (should be skipped)
|
||||
# Add a dependency with the same URI as dep1_uri to the destination store
|
||||
existing_dep1_data = b'{"data": "existing dependency 1 data", "deps": []}'
|
||||
existing_dep1_uri_in_dest = manager.add_raw(
|
||||
dep1_uri.asset_type, dep1_uri.asset_id, existing_dep1_data, "memory_deepcopy_dest"
|
||||
)
|
||||
self.assertEqual(existing_dep1_uri_in_dest.version, "2") # Should be version 2 now
|
||||
|
||||
# Deep copy the source asset again
|
||||
copied_uri_existing_dep = manager.deepcopy(
|
||||
src_uri, dest_store="memory_deepcopy_dest", store="memory_deepcopy_src"
|
||||
)
|
||||
# The top-level asset should be overwritten, incrementing its version
|
||||
self.assertEqual(copied_uri_existing_dep.version, "2")
|
||||
|
||||
# Verify the top-level asset was overwritten
|
||||
copied_asset_existing_dep = cast(
|
||||
MockAssetWithDeps, manager.get(copied_uri_existing_dep, store="memory_deepcopy_dest")
|
||||
)
|
||||
self.assertIsInstance(copied_asset_existing_dep, MockAssetWithDeps)
|
||||
self.assertEqual(
|
||||
copied_asset_existing_dep.get_data(),
|
||||
b'{"data": "source asset data", "deps": ["mock_asset_with_deps://dep1_id/1", "mock_asset_with_deps://dep2_id/1"]}',
|
||||
)
|
||||
|
||||
# Verify that the existing dependency was *not* overwritten
|
||||
retrieved_existing_dep1 = manager.get_raw(
|
||||
existing_dep1_uri_in_dest, store="memory_deepcopy_dest"
|
||||
)
|
||||
self.assertEqual(retrieved_existing_dep1, existing_dep1_data)
|
||||
|
||||
# Verify the dependencies in the copied asset still point to the correct
|
||||
# (existing) dependency in the destination store.
|
||||
copied_deps_existing_dep = copied_asset_existing_dep.get_dependencies()
|
||||
self.assertEqual(len(copied_deps_existing_dep), 2)
|
||||
self.assertIn(dep1_uri, copied_deps_existing_dep)
|
||||
self.assertIn(dep2_uri, copied_deps_existing_dep)
|
||||
|
||||
copied_dep1_existing = cast(MockAssetWithDeps, copied_deps_existing_dep[dep1_uri])
|
||||
self.assertIsInstance(copied_dep1_existing, MockAssetWithDeps)
|
||||
self.assertEqual(
|
||||
copied_dep1_existing.get_data(), b'{"data": "dependency 1 data", "deps": []}'
|
||||
) # Should be the original data from source
|
||||
|
||||
copied_dep2_existing = cast(MockAssetWithDeps, copied_deps_existing_dep[dep2_uri])
|
||||
self.assertIsInstance(copied_dep2_existing, MockAssetWithDeps)
|
||||
self.assertEqual(
|
||||
copied_dep2_existing.get_data(), b'{"data": "dependency 2 data", "deps": []}'
|
||||
) # Should be the newly copied raw data
|
||||
|
||||
# Test handling of existing top-level asset in the destination store (should be overwritten)
|
||||
# This was implicitly tested in the previous step where the top-level asset's
|
||||
# version was incremented. Let's add a more explicit test.
|
||||
overwrite_src_data = b'{"data": "overwrite source data", "deps": []}'
|
||||
overwrite_src_uri = manager.add_raw(
|
||||
MockAssetWithDeps.asset_type,
|
||||
"overwrite_source_id",
|
||||
overwrite_src_data,
|
||||
"memory_deepcopy_src",
|
||||
)
|
||||
|
||||
# Add an asset to the destination store with the same URI as overwrite_src_uri
|
||||
existing_dest_data = b'{"data": "existing destination data", "deps": []}'
|
||||
existing_dest_uri = manager.add_raw(
|
||||
overwrite_src_uri.asset_type,
|
||||
overwrite_src_uri.asset_id,
|
||||
existing_dest_data,
|
||||
"memory_deepcopy_dest",
|
||||
)
|
||||
self.assertEqual(existing_dest_uri.version, "1")
|
||||
|
||||
# Deep copy overwrite_src_uri to the existing destination URI
|
||||
copied_uri_overwrite_top = manager.deepcopy(
|
||||
overwrite_src_uri,
|
||||
dest_store="memory_deepcopy_dest",
|
||||
store="memory_deepcopy_src",
|
||||
dest=existing_dest_uri,
|
||||
)
|
||||
# The version should be incremented
|
||||
self.assertEqual(copied_uri_overwrite_top.version, "2")
|
||||
|
||||
# Verify the destination now contains the source data
|
||||
retrieved_overwritten_top = manager.get_raw(
|
||||
copied_uri_overwrite_top, store="memory_deepcopy_dest"
|
||||
)
|
||||
# Need to parse the data to get the actual content
|
||||
import json
|
||||
|
||||
retrieved_data_dict = json.loads(retrieved_overwritten_top.decode("utf-8"))
|
||||
self.assertEqual(retrieved_data_dict.get("data"), b"overwrite source data".decode("utf-8"))
|
||||
|
||||
# Test error handling for non-existent source asset
|
||||
non_existent_src_uri = AssetUri.build(MockAssetWithDeps.asset_type, "non_existent_src", "1")
|
||||
with self.assertRaises(FileNotFoundError) as cm:
|
||||
manager.deepcopy(
|
||||
non_existent_src_uri, dest_store="memory_deepcopy_dest", store="memory_deepcopy_src"
|
||||
)
|
||||
self.assertIn("Source asset", str(cm.exception))
|
||||
self.assertIn("not found", str(cm.exception))
|
||||
|
||||
# Test error handling for non-existent source store
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
manager.deepcopy(src_uri, dest_store="memory_deepcopy_dest", store="non_existent_store")
|
||||
self.assertIn("Source store", str(cm.exception))
|
||||
self.assertIn("not registered", str(cm.exception))
|
||||
|
||||
# Test error handling for non-existent destination store
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
manager.deepcopy(src_uri, dest_store="non_existent_store", store="memory_deepcopy_src")
|
||||
self.assertIn("Destination store", str(cm.exception))
|
||||
self.assertIn("not registered", str(cm.exception))
|
||||
|
||||
def test_exists(self):
|
||||
# Setup AssetManager with a MemoryStore
|
||||
memory_store = MemoryStore("memory_exists")
|
||||
manager = AssetManager()
|
||||
manager.register_store(memory_store)
|
||||
|
||||
# Create an asset
|
||||
test_uri = manager.add_raw("test_type", "test_id", b"data", "memory_exists")
|
||||
|
||||
# Test exists for an existing asset
|
||||
self.assertTrue(manager.exists(test_uri, store="memory_exists"))
|
||||
|
||||
# Test exists for a non-existent asset
|
||||
non_existent_uri = AssetUri.build("test_type", "non_existent_id", "1")
|
||||
self.assertFalse(manager.exists(non_existent_uri, store="memory_exists"))
|
||||
|
||||
# Test exists for a non-existent store (should raise ValueError)
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
manager.exists(test_uri, store="non_existent_store")
|
||||
self.assertIn("No store registered for name:", str(cm.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -62,7 +62,7 @@ class TestPathToolBit(PathTestWithAssets):
|
||||
|
||||
# Parameters should be loaded from the shape file and set on the tool bit's object
|
||||
self.assertEqual(bullnose_bit.obj.Diameter, FreeCAD.Units.Quantity("5.0 mm"))
|
||||
self.assertEqual(bullnose_bit.obj.FlatRadius, FreeCAD.Units.Quantity("1.5 mm"))
|
||||
self.assertEqual(bullnose_bit.obj.CornerRadius, FreeCAD.Units.Quantity("1.5 mm"))
|
||||
|
||||
def testToolBitPickle(self):
|
||||
"""Test if ToolBit is picklable"""
|
||||
|
||||
@@ -62,8 +62,8 @@ class TestToolBitBrowserWidget(PathTestWithAssets):
|
||||
search_term = "Endmill"
|
||||
self.widget._search_edit.setText(search_term)
|
||||
|
||||
# Directly trigger the fetch and filtering logic
|
||||
self.widget._trigger_fetch()
|
||||
# Directly trigger the filtering logic
|
||||
self.widget._update_list()
|
||||
|
||||
# Verify that the filter was applied to the list widget
|
||||
# We can check if items are hidden/shown based on the filter term
|
||||
@@ -100,28 +100,6 @@ class TestToolBitBrowserWidget(PathTestWithAssets):
|
||||
|
||||
self.assertEqual(actual_visible_uris, expected_visible_uris)
|
||||
|
||||
def test_lazy_loading_on_scroll(self):
|
||||
# This test requires more than self._batch_size toolbits to be effective.
|
||||
# The default test assets might not have enough.
|
||||
# We'll assume there are enough for the test structure.
|
||||
|
||||
initial_count = self.widget._tool_list_widget.count()
|
||||
if initial_count < self.widget._batch_size:
|
||||
self.skipTest("Not enough toolbits for lazy loading test.")
|
||||
|
||||
# Simulate scrolling to the bottom by emitting the signal
|
||||
scrollbar = self.widget._tool_list_widget.verticalScrollBar()
|
||||
# Set the scrollbar value to its maximum to simulate reaching the end
|
||||
scrollbar.valueChanged.emit(scrollbar.maximum())
|
||||
|
||||
# Verify that more items were loaded
|
||||
new_count = self.widget._tool_list_widget.count()
|
||||
self.assertGreater(new_count, initial_count)
|
||||
# Verify that the number of new items is approximately the batch size
|
||||
self.assertAlmostEqual(
|
||||
new_count - initial_count, self.widget._batch_size, delta=5
|
||||
) # Allow small delta
|
||||
|
||||
def test_tool_selected_signal(self):
|
||||
mock_slot = MagicMock()
|
||||
self.widget.toolSelected.connect(mock_slot)
|
||||
|
||||
@@ -22,7 +22,9 @@
|
||||
|
||||
"""Unit tests for the ToolBitListWidget."""
|
||||
|
||||
from typing import cast
|
||||
import unittest
|
||||
from Path.Tool.toolbit import ToolBit
|
||||
from Path.Tool.toolbit.ui.toollist import ToolBitListWidget, ToolBitUriRole
|
||||
from Path.Tool.toolbit.ui.tablecell import TwoLineTableCell
|
||||
from .PathTestUtils import PathTestWithAssets # Import the base test class
|
||||
@@ -37,7 +39,7 @@ class TestToolBitListWidget(PathTestWithAssets):
|
||||
|
||||
def test_add_toolbit(self):
|
||||
# Get a real ToolBit asset
|
||||
toolbit = self.assets.get("toolbit://5mm_Endmill")
|
||||
toolbit = cast(ToolBit, self.assets.get("toolbit://5mm_Endmill"))
|
||||
tool_no = 1
|
||||
|
||||
self.widget.add_toolbit(toolbit, str(tool_no))
|
||||
@@ -61,8 +63,8 @@ class TestToolBitListWidget(PathTestWithAssets):
|
||||
|
||||
def test_clear_list(self):
|
||||
# Add some real items first
|
||||
toolbit1 = self.assets.get("toolbit://5mm_Endmill")
|
||||
toolbit2 = self.assets.get("toolbit://slittingsaw")
|
||||
toolbit1 = cast(ToolBit, self.assets.get("toolbit://5mm_Endmill"))
|
||||
toolbit2 = cast(ToolBit, self.assets.get("toolbit://slittingsaw"))
|
||||
self.widget.add_toolbit(toolbit1, 1)
|
||||
self.widget.add_toolbit(toolbit2, 2)
|
||||
self.assertEqual(self.widget.count(), 2)
|
||||
@@ -72,9 +74,9 @@ class TestToolBitListWidget(PathTestWithAssets):
|
||||
|
||||
def test_apply_filter(self):
|
||||
# Add items with distinct text for filtering
|
||||
toolbit1 = self.assets.get("toolbit://5mm_Endmill")
|
||||
toolbit2 = self.assets.get("toolbit://slittingsaw")
|
||||
toolbit3 = self.assets.get("toolbit://probe")
|
||||
toolbit1 = cast(ToolBit, self.assets.get("toolbit://5mm_Endmill"))
|
||||
toolbit2 = cast(ToolBit, self.assets.get("toolbit://slittingsaw"))
|
||||
toolbit3 = cast(ToolBit, self.assets.get("toolbit://probe"))
|
||||
|
||||
self.widget.add_toolbit(toolbit1, 1)
|
||||
self.widget.add_toolbit(toolbit2, 2)
|
||||
@@ -117,8 +119,8 @@ class TestToolBitListWidget(PathTestWithAssets):
|
||||
self.assertEqual(cell.search_highlight, "3mm")
|
||||
|
||||
def test_get_selected_toolbit_uri(self):
|
||||
toolbit1 = self.assets.get("toolbit://5mm_Endmill")
|
||||
toolbit2 = self.assets.get("toolbit://slittingsaw")
|
||||
toolbit1 = cast(ToolBit, self.assets.get("toolbit://5mm_Endmill"))
|
||||
toolbit2 = cast(ToolBit, self.assets.get("toolbit://slittingsaw"))
|
||||
|
||||
self.widget.add_toolbit(toolbit1, 1)
|
||||
self.widget.add_toolbit(toolbit2, 2)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import yaml
|
||||
import json
|
||||
from typing import Type, cast
|
||||
import FreeCAD
|
||||
@@ -6,6 +7,7 @@ from Path.Tool.toolbit import ToolBit, ToolBitEndmill
|
||||
from Path.Tool.toolbit.serializers import (
|
||||
FCTBSerializer,
|
||||
CamoticsToolBitSerializer,
|
||||
YamlToolBitSerializer,
|
||||
)
|
||||
from Path.Tool.assets.asset import Asset
|
||||
from Path.Tool.assets.serializer import AssetSerializer
|
||||
@@ -132,3 +134,72 @@ class TestFCTBSerializer(_BaseToolBitSerializerTestCase):
|
||||
self.assertEqual(deserialized_bit.get_shape_name(), "Endmill")
|
||||
self.assertEqual(str(deserialized_bit.get_diameter()), "4.12 mm")
|
||||
self.assertEqual(str(deserialized_bit.get_length()), "15.0 mm")
|
||||
|
||||
|
||||
class TestYamlToolBitSerializer(_BaseToolBitSerializerTestCase):
|
||||
serializer_class = YamlToolBitSerializer
|
||||
|
||||
def test_serialize(self):
|
||||
super().test_serialize()
|
||||
serialized_data = self.serializer_class.serialize(self.test_tool_bit)
|
||||
# YAML specific assertions
|
||||
data = yaml.safe_load(serialized_data.decode("utf-8"))
|
||||
self.assertEqual(data.get("id"), "5mm_Endmill")
|
||||
self.assertEqual(data.get("name"), "Test Tool")
|
||||
self.assertEqual(data.get("shape"), "endmill.fcstd")
|
||||
self.assertEqual(data.get("shape-type"), "Endmill")
|
||||
self.assertEqual(data.get("parameter", {}).get("Diameter"), "4.12 mm")
|
||||
self.assertEqual(data.get("parameter", {}).get("Length"), "15.00 mm")
|
||||
|
||||
def test_extract_dependencies(self):
|
||||
"""Test dependency extraction for YAML."""
|
||||
yaml_data = (
|
||||
b"name: Test Tool\n"
|
||||
b"shape: endmill\n"
|
||||
b"shape-type: Endmill\n"
|
||||
b"parameter:\n"
|
||||
b" Diameter: 4.12 mm\n"
|
||||
b" Length: 15.0 mm\n"
|
||||
b"attribute: {}\n"
|
||||
)
|
||||
dependencies = self.serializer_class.extract_dependencies(yaml_data)
|
||||
self.assertIsInstance(dependencies, list)
|
||||
self.assertEqual(len(dependencies), 1)
|
||||
self.assertEqual(dependencies[0], AssetUri.build("toolbitshape", "endmill"))
|
||||
|
||||
def test_deserialize(self):
|
||||
# Create a known serialized data string based on the YAML format
|
||||
yaml_data = (
|
||||
b"id: TestID\n"
|
||||
b"name: Test Tool\n"
|
||||
b"shape: endmill\n"
|
||||
b"shape-type: Endmill\n"
|
||||
b"parameter:\n"
|
||||
b" Diameter: 4.12 mm\n"
|
||||
b" Length: 15.0 mm\n"
|
||||
b"attribute: {}\n"
|
||||
)
|
||||
# Create a ToolBitShapeEndmill instance for 'endmill'
|
||||
shape = ToolBitShapeEndmill("endmill")
|
||||
|
||||
# Create the dependencies dictionary with the shape instance
|
||||
dependencies: Mapping[AssetUri, Asset] = {AssetUri.build("toolbitshape", "endmill"): shape}
|
||||
|
||||
# Provide dummy id and dependencies for deserialization test
|
||||
deserialized_bit = cast(
|
||||
ToolBitEndmill,
|
||||
self.serializer_class.deserialize(yaml_data, "TestID", dependencies=dependencies),
|
||||
)
|
||||
self.assertIsInstance(deserialized_bit, ToolBit)
|
||||
self.assertEqual(deserialized_bit.id, "TestID")
|
||||
self.assertEqual(deserialized_bit.label, "Test Tool")
|
||||
self.assertEqual(deserialized_bit.get_shape_name(), "Endmill")
|
||||
self.assertEqual(str(deserialized_bit.get_diameter()), "4.12 mm")
|
||||
self.assertEqual(str(deserialized_bit.get_length()), "15.0 mm")
|
||||
|
||||
# Test with ID argument.
|
||||
deserialized_bit = cast(
|
||||
ToolBitEndmill,
|
||||
self.serializer_class.deserialize(yaml_data, id="test_id", dependencies=dependencies),
|
||||
)
|
||||
self.assertEqual(deserialized_bit.id, "test_id")
|
||||
|
||||
@@ -144,13 +144,13 @@ class TestLinuxCNCLibrarySerializer(TestPathToolLibrarySerializerBase):
|
||||
# Verify the content format (basic check)
|
||||
lines = serialized_data.decode("ascii", "ignore").strip().split("\n")
|
||||
self.assertEqual(len(lines), 3)
|
||||
self.assertEqual(lines[0], "T1 P0 D6.000 ;Endmill 6mm")
|
||||
self.assertEqual(lines[1], "T2 P0 D3.000 ;Endmill 3mm")
|
||||
self.assertEqual(lines[2], "T3 P0 D5.000 ;Ballend 5mm")
|
||||
self.assertEqual(lines[0], "T1 P0 X0 Y0 Z0 A0 B0 C0 U0 V0 W0 D6.00 I0 J0 Q0 ;Endmill 6mm")
|
||||
self.assertEqual(lines[1], "T2 P0 X0 Y0 Z0 A0 B0 C0 U0 V0 W0 D3.00 I0 J0 Q0 ;Endmill 3mm")
|
||||
self.assertEqual(lines[2], "T3 P0 X0 Y0 Z0 A0 B0 C0 U0 V0 W0 D5.00 I0 J0 Q0 ;Ballend 5mm")
|
||||
|
||||
def test_linuxcnc_deserialize_not_implemented(self):
|
||||
serializer = LinuxCNCSerializer
|
||||
dummy_data = b"T1 D6.0 ;Endmill 6mm\n"
|
||||
dummy_data = b"T1 P0 X0 Y0 Z0 A0 B0 C0 U0 V0 W0 D6.00 I0 J0 Q0 ;Endmill 6mm\n"
|
||||
with self.assertRaises(NotImplementedError):
|
||||
serializer.deserialize(dummy_data, "dummy_id", {})
|
||||
|
||||
|
||||
@@ -153,8 +153,8 @@ class TestPathToolShapeClasses(PathTestWithAssets):
|
||||
self.assertEqual(ToolBitShape.resolve_name("ballend").asset_id, "ballend")
|
||||
self.assertEqual(ToolBitShape.resolve_name("v-bit").asset_id, "v-bit")
|
||||
self.assertEqual(ToolBitShape.resolve_name("vbit").asset_id, "vbit")
|
||||
self.assertEqual(ToolBitShape.resolve_name("torus").asset_id, "torus")
|
||||
self.assertEqual(ToolBitShape.resolve_name("torus.fcstd").asset_id, "torus")
|
||||
self.assertEqual(ToolBitShape.resolve_name("bullnose").asset_id, "bullnose")
|
||||
self.assertEqual(ToolBitShape.resolve_name("bullnose.fcstd").asset_id, "bullnose")
|
||||
self.assertEqual(ToolBitShape.resolve_name("SlittingSaw").asset_id, "SlittingSaw")
|
||||
# Test unknown name - should return the input name
|
||||
self.assertEqual(ToolBitShape.resolve_name("nonexistent").asset_id, "nonexistent")
|
||||
@@ -336,12 +336,12 @@ class TestPathToolShapeClasses(PathTestWithAssets):
|
||||
shape = self._test_shape_common("bullnose")
|
||||
self.assertEqual(shape["Diameter"].Value, 5.0)
|
||||
self.assertEqual(unit(shape["Diameter"]), "mm")
|
||||
self.assertEqual(shape["FlatRadius"].Value, 1.5)
|
||||
self.assertEqual(unit(shape["FlatRadius"]), "mm")
|
||||
self.assertEqual(shape["CornerRadius"].Value, 1.5)
|
||||
self.assertEqual(unit(shape["CornerRadius"]), "mm")
|
||||
# Need an instance to get parameter labels, get it from the asset manager
|
||||
uri = ToolBitShape.resolve_name("bullnose")
|
||||
instance = self.assets.get(uri)
|
||||
self.assertEqual(instance.get_parameter_label("FlatRadius"), "Torus radius")
|
||||
self.assertEqual(instance.get_parameter_label("CornerRadius"), "Corner radius")
|
||||
|
||||
def test_toolbitshapevbit_defaults(self):
|
||||
"""Test ToolBitShapeVBit default parameters and labels."""
|
||||
|
||||
Reference in New Issue
Block a user