Skip to content

Commit

Permalink
wired up the rest of the ignored MGraph__Json tests
Browse files Browse the repository at this point in the history
improved test coverage
  • Loading branch information
DinisCruz committed Jan 15, 2025
1 parent e8de22f commit 74edd44
Show file tree
Hide file tree
Showing 21 changed files with 419 additions and 233 deletions.
3 changes: 3 additions & 0 deletions mgraph_ai/mgraph/domain/Domain__MGraph__Graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,6 @@ def node(self, node_id: Random_Guid) -> Domain__MGraph__Node:
def nodes(self) -> List[Domain__MGraph__Node]:
return [self.mgraph_node(node=node) for node in self.model.nodes()]

def nodes_ids(self):
return self.model.nodes_ids()

4 changes: 2 additions & 2 deletions mgraph_ai/mgraph/domain/Domain__MGraph__Node.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ def add_node(self, node: Model__MGraph__Node) -> None:
def models__edges(self) -> List[Model__MGraph__Edge]: # Get all model edges connected to this node
connected_edges = []
for edge in self.graph.edges():
if edge.from_node_id() == self.node_id() or edge.to_node_id() == self.node_id():
if edge.from_node_id() == self.node_id or edge.to_node_id() == self.node_id:
connected_edges.append(edge)
return connected_edges

Expand All @@ -33,7 +33,7 @@ def models__from_edges(self) -> List[Model__MGraph__Edge]:
def models__to_edges(self) -> List[Model__MGraph__Edge]: # Get model edges where this node is the target
incoming_edges = []
for edge in self.graph.edges():
if edge.to_node_id() == self.node_id():
if edge.to_node_id() == self.node_id:
incoming_edges.append(edge)
return incoming_edges

Expand Down
3 changes: 0 additions & 3 deletions mgraph_ai/mgraph/utils/MGraph__Random_Graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,9 +61,6 @@ def create_random_graph(self, num_nodes: int = 10, num_edges: Optional[int] = No

return self.graph

def empty_graph(self):
return self.setup().graph

def create_random_mgraph(num_nodes=2, num_edges=2) -> MGraph: # Create an empty graph with no nodes or edges
return MGraph__Random_Graph().setup().create_random_graph(num_nodes=num_nodes, num_edges=num_edges)

Expand Down
8 changes: 0 additions & 8 deletions mgraph_ai/mgraph/utils/MGraph__Static__Graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,6 @@ class MGraph__Static__Graph(Type_Safe):
node_ids: List[Random_Guid]
edge_ids: List[Random_Guid]

@property
def total_nodes(self) -> int: # Returns total number of nodes in the graph
return len(self.node_ids)

@property
def total_edges(self) -> int: # Returns total number of edges in the graph
return len(self.edge_ids)

def create_nodes(self, count: int) -> List[Random_Guid]: # Creates specified number of nodes
with self.graph.edit() as edit:
return [edit.new_node().node_id for _ in range(count)]
Expand Down
77 changes: 39 additions & 38 deletions mgraph_ai/providers/json/domain/Domain__MGraph__Json__Graph.py
Original file line number Diff line number Diff line change
@@ -1,41 +1,44 @@
from typing import Any, Optional
from mgraph_ai.mgraph.domain.Domain__MGraph__Graph import Domain__MGraph__Graph
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Types import Domain__MGraph__Json__Types
from mgraph_ai.providers.json.models.Model__MGraph__Json__Graph import Model__MGraph__Json__Graph
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node import Domain__MGraph__Json__Node
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node__Dict import Domain__MGraph__Json__Node__Dict
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node__List import Domain__MGraph__Json__Node__List
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node__Value import Domain__MGraph__Json__Node__Value
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node import Schema__MGraph__Json__Node
from typing import Any, Optional
from mgraph_ai.providers.json.models.Model__MGraph__Json__Node__Dict import Model__MGraph__Json__Node__Dict
from mgraph_ai.providers.json.models.Model__MGraph__Json__Node__List import Model__MGraph__Json__Node__List
from mgraph_ai.mgraph.domain.Domain__MGraph__Graph import Domain__MGraph__Graph
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Types import Domain__MGraph__Json__Types
from mgraph_ai.providers.json.models.Model__MGraph__Json__Graph import Model__MGraph__Json__Graph
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node import Domain__MGraph__Json__Node
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node__Dict import Domain__MGraph__Json__Node__Dict
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node__List import Domain__MGraph__Json__Node__List
from mgraph_ai.providers.json.domain.Domain__MGraph__Json__Node__Value import Domain__MGraph__Json__Node__Value
from mgraph_ai.providers.json.models.Model__MGraph__Json__Node__Value import Model__MGraph__Json__Node__Value
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node import Schema__MGraph__Json__Node
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node__Dict import Schema__MGraph__Json__Node__Dict
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node__List import Schema__MGraph__Json__Node__List
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node__Value import Schema__MGraph__Json__Node__Value
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node__Value__Data import Schema__MGraph__Json__Node__Value__Data


class Domain__MGraph__Json__Graph(Domain__MGraph__Graph):
domain_types : Domain__MGraph__Json__Types
model : Model__MGraph__Json__Graph

def root(self) -> Domain__MGraph__Json__Node: # Get root node
"""Get the root node, creating it if it doesn't exist"""
def root(self) -> Domain__MGraph__Json__Node: #Get the root node, creating it if it doesn't exist
if not self.model.data.graph_data.root_id:
schema_node = Schema__MGraph__Json__Node() # Create basic root node
node = self.model.add_node(schema_node)
self.model.data.graph_data.root_id = node.node_id
return Domain__MGraph__Json__Node(node=node, graph=self.model)

root_model = self.model.node(self.model.data.graph_data.root_id)
if not root_model:
raise ValueError("Root node ID exists but node not found")

return Domain__MGraph__Json__Node(node=root_model, graph=self.model)

def root_content(self) -> Optional[Domain__MGraph__Json__Node]: # Get the content node
"""Get the content node attached to root (if any)"""
def root_content(self) -> Optional[Domain__MGraph__Json__Node]: # Get the content node attached to root (if any)"""
root = self.root()
edges = self.model.node__from_edges(root.node_id)
if not edges:
return None

content_node = self.model.node(edges[0].to_node_id()) # Get first child

edge__to_node_id = edges[0].to_node_id() # first edge
content_node = self.model.node(edge__to_node_id) # Get first child
# Return appropriate domain node type
if isinstance(content_node.data, Schema__MGraph__Json__Node__Dict):
return Domain__MGraph__Json__Node__Dict(node=content_node, graph=self.model)
Expand All @@ -44,20 +47,15 @@ def root_content(self) -> Optional[Domain__MGraph__Json__Node]:
elif isinstance(content_node.data, Schema__MGraph__Json__Node__Value):
return Domain__MGraph__Json__Node__Value(node=content_node, graph=self.model)

return None

def set_root_content(self, data: Any) -> Domain__MGraph__Json__Node: # Set content node
"""Set the JSON content, creating appropriate node type"""
def set_root_content(self, data: Any) -> Domain__MGraph__Json__Node: # Set the JSON content, creating appropriate node type
root = self.root()

# Remove any existing content
edges = self.model.node__from_edges(root.node_id)
edges = self.model.node__from_edges(root.node_id) # Remove any existing content
for edge in edges:
self.model.delete_edge(edge.edge_id())
self.model.delete_node(edge.to_node_id())

# Create appropriate node type for new content
if isinstance(data, dict):
if isinstance(data, dict): # Create appropriate node type for new content
content = self.new_dict_node(data)
elif isinstance(data, (list, tuple)):
content = self.new_list_node(data)
Expand All @@ -68,31 +66,34 @@ def set_root_content(self, data: Any) -> Domain__MGraph__Json__Node:
self.model.new_edge(from_node_id=root.node_id, to_node_id=content.node_id)
return content

def new_dict_node(self, properties=None) -> Domain__MGraph__Json__Node__Dict: # Create dictionary node
"""Create a new dictionary node with optional initial properties"""
def new_dict_node(self, properties=None) -> Domain__MGraph__Json__Node__Dict: # Create a new dictionary node with optional initial properties"""
schema_node = Schema__MGraph__Json__Node__Dict()
node = self.model.add_node(schema_node)
dict_node = Domain__MGraph__Json__Node__Dict(node=node, graph=self.model)
#node = self.model.add_node(schema_node) # todo:: find way to use self.model method
self.model.data.nodes[schema_node.node_id] = schema_node # so that we don't need to add this here
model_node = Model__MGraph__Json__Node__Dict(data=schema_node)
dict_node = Domain__MGraph__Json__Node__Dict(node=model_node, graph=self.model)

if properties:
dict_node.update(properties)

return dict_node

def new_list_node(self, items=None) -> Domain__MGraph__Json__Node__List: # Create list node
"""Create a new list node with optional initial items"""
def new_list_node(self, items=None) -> Domain__MGraph__Json__Node__List: # Create a new list node with optional initial items"""
schema_node = Schema__MGraph__Json__Node__List()
node = self.model.add_node(schema_node)
list_node = Domain__MGraph__Json__Node__List(node=node, graph=self.model)
#node = self.model.add_node(schema_node) # todo:: find way to use self.model method
self.model.data.nodes[schema_node.node_id] = schema_node # so that we don't need to add this here
model_node = Model__MGraph__Json__Node__List(data=schema_node)
list_node = Domain__MGraph__Json__Node__List(node=model_node, graph=self.model)

if items:
list_node.extend(items)

return list_node

def new_value_node(self, value: Any) -> Domain__MGraph__Json__Node__Value: # Create value node
"""Create a new value node with the given value"""
node_data = Schema__MGraph__Json__Node__Value__Data(value=value, value_type=type(value))
def new_value_node(self, value: Any) -> Domain__MGraph__Json__Node__Value: # Create a new value node with the given value
node_data = Schema__MGraph__Json__Node__Value__Data(value=value, value_type=type(value))
schema_node = Schema__MGraph__Json__Node__Value(node_data=node_data)
node = self.model.add_node(schema_node)
return Domain__MGraph__Json__Node__Value(node=node, graph=self.model)
#node = self.model.add_node(schema_node) # todo:: find way to use self.model method
model_node = Model__MGraph__Json__Node__Value(data=schema_node) # so that we don't need to add this here
self.model.data.nodes[schema_node.node_id] = schema_node
return Domain__MGraph__Json__Node__Value(node=model_node, graph=self.model)
18 changes: 1 addition & 17 deletions mgraph_ai/providers/json/domain/Domain__MGraph__Json__Node.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,5 @@
from typing import Optional
from mgraph_ai.mgraph.domain.Domain__MGraph__Node import Domain__MGraph__Node
from mgraph_ai.providers.json.models.Model__MGraph__Json__Node import Model__MGraph__Json__Node
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node__Dict import Schema__MGraph__Json__Node__Dict
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node__List import Schema__MGraph__Json__Node__List
from mgraph_ai.providers.json.schemas.Schema__MGraph__Json__Node__Value import Schema__MGraph__Json__Node__Value


class Domain__MGraph__Json__Node(Domain__MGraph__Node):
node: Model__MGraph__Json__Node # Reference to node model

def get_type(self) -> Optional[str]: # Get node type
"""Get the type of this JSON node (dict, list, or value)"""
node_type = self.node.data.node_type
if node_type == Schema__MGraph__Json__Node__Dict:
return 'dict'
elif node_type == Schema__MGraph__Json__Node__List:
return 'list'
elif node_type == Schema__MGraph__Json__Node__Value:
return 'value'
return None
node: Model__MGraph__Json__Node # Reference to node model
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,6 @@ def properties(self) -> Dict[str, Any]:
result = {}
for edge in self.models__from_edges():
property_node = self.model__node_from_edge(edge)
if property_node.data.node_type != Schema__MGraph__Json__Node__Property:
continue

for value_edge in self.graph.edges(): # todo: see why we need to use self.graph.edges()
if value_edge.from_node_id() == property_node.node_id:
Expand Down
Loading

0 comments on commit 74edd44

Please sign in to comment.