Skip to content

Commit

Permalink
Add parsing tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@john681611
john681611 committed Sep 15, 2023
1 parent bd709bc commit 3822a51
Show file tree
Hide file tree
Showing 2 changed files with 185 additions and 16 deletions.
46 changes: 30 additions & 16 deletions application/database/db.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -276,6 +276,20 @@ def add_dbnode(self, dbnode: Node):
database_="neo4j",
)
return
if dbnode.ntype == "Code":
self.driver.execute_query(
"MERGE (n:Code {id: $nid, name: $name, section: $section, sectionID: $sectionID, subsection: $subsection, tags: $tags, version: $version, description: $description, doctype: $doctype, links: $links, metadata: $metadata, hyperlink: $hyperlink})",
name=dbnode.name,
doctype=dbnode.ntype,
nid=dbnode.id,
description=dbnode.description,
links=[], # dbnode.links,
tags=dbnode.tags,
metadata="{}", # dbnode.metadata,
hyperlink="", # dbnode.hyperlink or "",
version=dbnode.version or "",
)
return
raise Exception(f"Unknown DB type: {dbnode.ntype}")

@classmethod
Expand Down Expand Up @@ -383,29 +397,29 @@ def parse_node(self, node: neo4j.graph.Node) -> cre_defs.Document:
name = node["name"]
id = node["id"] if "id" in node else None
description = node["description"] if "description" in node else None
links = [self.parse_link(link) for link in node["links"]]
# links = [self.parse_link(link) for link in node["links"]]
tags = node["tags"]
metadata = node["metadata"]
# metadata = node["metadata"]
if "Code" in node.labels:
return cre_defs.Code(
name=name,
id=id,
description=description,
links=links,
# links=links,
tags=tags,
metadata=metadata,
hyperlink=(node["hyperlink"] if "hyperlink" in node else None),
# metadata=metadata,
# hyperlink=(node["hyperlink"] if "hyperlink" in node else None),
version=(node["version"] if "version" in node else None),
)
if "Standard" in node.labels:
return cre_defs.Standard(
name=name,
id=id,
description=description,
links=links,
# links=links,
tags=tags,
metadata=metadata,
hyperlink=(node["hyperlink"] if "hyperlink" in node else None),
# metadata=metadata,
# hyperlink=(node["hyperlink"] if "hyperlink" in node else None),
version=(node["version"] if "version" in node else None),
section=node["section"],
sectionID=node["sectionID"],
Expand All @@ -416,10 +430,10 @@ def parse_node(self, node: neo4j.graph.Node) -> cre_defs.Document:
name=name,
id=id,
description=description,
links=links,
# links=links,
tags=tags,
metadata=metadata,
hyperlink=(node["hyperlink"] if "hyperlink" in node else None),
# metadata=metadata,
# hyperlink=(node["hyperlink"] if "hyperlink" in node else None),
version=(node["version"] if "version" in node else None),
section=node["section"],
sectionID=node["sectionID"],
Expand All @@ -430,15 +444,15 @@ def parse_node(self, node: neo4j.graph.Node) -> cre_defs.Document:
name=name,
id=id,
description=description,
links=links,
# links=links,
tags=tags,
metadata=metadata,
# metadata=metadata,
)
raise Exception(f"Unknown node {node.labels}")

@classmethod
def parse_link(self, link):
return cre_defs.Link(ltype=link["ltype"], tags=link["tags"])
# @classmethod
# def parse_link(self, link):
# return cre_defs.Link(ltype=link["ltype"], tags=link["tags"])


class CRE_Graph:
Expand Down
155 changes: 155 additions & 0 deletions application/tests/db_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from pprint import pprint
from pydoc import doc
from typing import Any, Dict, List, Union
import neo4j

import yaml
from application import create_app, sqla # type: ignore
Expand Down Expand Up @@ -1308,6 +1309,160 @@ def test_gap_analysis_duplicate_link_path_existing_higher(self, gap_mock):
}
self.assertEqual(collection.gap_analysis(["a", "b"]), expected)

def test_neo_db_parse_node_code(self):
collection = db.Node_collection()
name = "name"
id = "id"
description = "description"
tags = "tags"
version = "version"
expected = defs.Code(
name=name,
id=id,
description=description,
tags=tags,
version=version,
)
graph_node = neo4j.graph.Node(
None,
"123",
"id",
n_labels=["Code"],
properties={
"name": name,
"id": id,
"description": description,
"tags": tags,
"version": version,
},
)
self.assertEqual(collection.neo_db.parse_node(graph_node), expected)

def test_neo_db_parse_node_standard(self):
collection = db.Node_collection()
name = "name"
id = "id"
description = "description"
tags = "tags"
version = "version"
section = "section"
sectionID = "sectionID"
subsection = "subsection"
expected = defs.Standard(
name=name,
id=id,
description=description,
tags=tags,
version=version,
section=section,
sectionID=sectionID,
subsection=subsection,
)
graph_node = neo4j.graph.Node(
None,
"123",
"id",
n_labels=["Standard"],
properties={
"name": name,
"id": id,
"description": description,
"tags": tags,
"version": version,
"section": section,
"sectionID": sectionID,
"subsection": subsection,
},
)
self.assertEqual(collection.neo_db.parse_node(graph_node), expected)

def test_neo_db_parse_node_tool(self):
collection = db.Node_collection()
name = "name"
id = "id"
description = "description"
tags = "tags"
version = "version"
section = "section"
sectionID = "sectionID"
subsection = "subsection"
expected = defs.Tool(
name=name,
id=id,
description=description,
tags=tags,
version=version,
section=section,
sectionID=sectionID,
subsection=subsection,
)
graph_node = neo4j.graph.Node(
None,
"123",
"id",
n_labels=["Tool"],
properties={
"name": name,
"id": id,
"description": description,
"tags": tags,
"version": version,
"section": section,
"sectionID": sectionID,
"subsection": subsection,
},
)
self.assertEqual(collection.neo_db.parse_node(graph_node), expected)

def test_neo_db_parse_node_cre(self):
collection = db.Node_collection()
name = "name"
id = "id"
description = "description"
tags = "tags"
expected = defs.CRE(
name=name,
id=id,
description=description,
tags=tags,
)
graph_node = neo4j.graph.Node(
None,
"123",
"id",
n_labels=["CRE"],
properties={
"name": name,
"id": id,
"description": description,
"tags": tags,
},
)
self.assertEqual(collection.neo_db.parse_node(graph_node), expected)

def test_neo_db_parse_node_unknown(self):
collection = db.Node_collection()
name = "name"
id = "id"
description = "description"
tags = "tags"
graph_node = neo4j.graph.Node(
None,
"123",
"id",
n_labels=["ABC"],
properties={
"name": name,
"id": id,
"description": description,
"tags": tags,
},
)
with self.assertRaises(Exception) as cm:
collection.neo_db.parse_node(graph_node)

self.assertEqual(str(cm.exception), "Unknown node frozenset({'ABC'})")

def test_get_embeddings_by_doc_type_paginated(self):
"""Given: a range of embedding for Nodes and a range of embeddings for CREs
when called with doc_type CRE return the cre embeddings
Expand Down

0 comments on commit 3822a51

Please sign in to comment.