infeeeee/dyn2py
1
0
Fork 0
mirror of https://github.com/infeeeee/dyn2py synced 2025-12-16 22:16:18 +01:00
Code Issues Packages Projects Releases Activity

More tests, refactor PythonNode

Browse Source
This commit is contained in:
Péter Gyetvai
2023-03-13 21:02:38 +01:00
parent c40ff88bcf
commit 7ef6a0f886
6 changed files with 148 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
TODO.md
View File

@@ -6,7 +6,7 @@
- [x] File
- [ ] DynamoFile
- [ ] PythonFile
- [x] PythonNode
- [ ] PythonNode
- [ ] run()
## CI/CD

3
dyn2py/exceptions.py
View File

@@ -5,6 +5,9 @@ class DynamoFileException(Exception):
class PythonNodeNotFoundException(Exception):
pass
class PythonNodeException(Exception):
pass
class PythonFileException(Exception):
pass

95
dyn2py/files.py
View File

@@ -25,7 +25,7 @@ class File():
Args:
filepath (pathlib.Path | str): Path to the python file or Dynamo graph
read_from_disk (bool, optional): Read the file from disk. Useful for new PythonFiles. Defaults to True.
read_from_disk (bool, optional): Read the file from disk. Set to false to get only metadata. Defaults to True.
"""
self.filepath: pathlib.Path
@@ -57,7 +57,8 @@ class File():
if self.is_dynamo_file():
self.__class__ = DynamoFile
# Always read DynamoFiles, they should exist:
# Read DynamoFiles, they should exist:
if read_from_disk:
self.read_file()
elif self.is_python_file():
@@ -108,18 +109,21 @@ class File():
"""
return bool(self.extension == ".py")
def write(self, options: Options) -> None:
def write(self, options: Options | None = None) -> None:
"""Prepare writing file to the disk:
create backup, process dry-run, call filetype specific write_file() methods
Should be called on subclasses!
Args:
options(Options): Run options.
options(Options | None, optional): Run options. Defaults to None.
Raises:
TypeError: If called on a File object
"""
if not options:
options = Options()
# This should only work on subclasses:
if type(self).__name__ == "File":
raise TypeError("This method shouldn't be called on File objects!")
@@ -249,8 +253,7 @@ class DynamoFile(File):
for p_node in full_python_nodes:
python_node = PythonNode(
node_dict_from_dyn=p_node,
full_nodeviews_dict_from_dyn=node_views,
source_dynamo_file=self)
dynamo_file=self)
self.python_nodes.add(python_node)
def get_python_node_by_id(self, node_id: str) -> "PythonNode":
@@ -329,7 +332,7 @@ class DynamoFile(File):
python_folder = self.dirpath
python_files_in_folder = [PythonFile(f) for f in python_folder.iterdir()
if File(f).is_python_file()]
if File(f, read_from_disk=False).is_python_file()]
related_python_files = [
p for p in python_files_in_folder if p.get_source_dynamo_file().uuid == self.uuid]
@@ -495,12 +498,7 @@ class PythonFile(File):
if not dynamo_file:
dynamo_file = self.get_source_dynamo_file()
new_python_node = PythonNode(
node_id=self.header_data["py_id"],
engine=self.header_data["py_engine"],
code=self.code,
checksum=hashlib.md5(self.code.encode()).hexdigest()
)
new_python_node = PythonNode(python_file=self)
old_python_node = dynamo_file.get_python_node_by_id(
self.header_data["py_id"])
@@ -562,46 +560,42 @@ class PythonNode():
filename: pathlib.Path | str
"""The filename the node should be saved as, including the .py extension"""
filepath: pathlib.Path
"""The path is shoul"""
def __init__(self, node_dict_from_dyn: dict = {}, full_nodeviews_dict_from_dyn: dict = {},
node_id: str = "", engine: str = "IronPython2", code: str = "", checksum: str = "", name: str = "",
source_dynamo_file: DynamoFile | None = None) -> None:
"""A PythonNode object. If node_dict_view is given, string parameters are ignored.
def __init__(self,
node_dict_from_dyn: dict = {},
dynamo_file: DynamoFile | None = None,
python_file: PythonFile | None = None,
) -> None:
"""A PythonNode object. Add dict and dynamo_file, or only python_file.
Args:
node_dict_from_dyn (dict, optional): The dict of the node from a dyn file. Defaults to {}.
full_nodeviews_dict_from_dyn (dict, optional): The full nodeviews dict from a dyn file. Defaults to {}.
node_id (str, optional): Id of the node. Defaults to "".
engine (str, optional): Engine of the node. Defaults to "".
code (str, optional): The code text. Defaults to "".
checksum (str, optional): Checksum of the code . Defaults to "".
name (str, optional): The name of the node. Defaults to "".
source_dynamo_file (DynamoFile, optional): The file the node is from, to generate filename and filepath. Defaults to None.
dynamo_file (DynamoFile, optional): The file the node is from. Defaults to None.
python_file (PythonFile, optional): The python file to be converted to node. Defaults to None.
Raises:
PythonNodeException: Wrong arguments were given
"""
if node_dict_from_dyn:
# Initialize from dynamo file:
if node_dict_from_dyn and dynamo_file and not python_file:
self.id = node_dict_from_dyn["Id"]
# Older dynamo files doesn't have "Engine" property, fall back to the default
# Older dynamo files doesn't have "Engine" property, fall back to IronPython2
if "Engine" in node_dict_from_dyn:
self.engine = node_dict_from_dyn["Engine"]
else:
self.engine = engine
self.code = node_dict_from_dyn["Code"]
self.checksum = hashlib.md5(self.code.encode()).hexdigest()
if full_nodeviews_dict_from_dyn:
self.name = next(
(v["Name"] for v in full_nodeviews_dict_from_dyn if v["Id"] == node_dict_from_dyn["Id"]), "")
else:
self.name = name
else:
self.id = node_id
self.engine = engine
self.code = code
self.checksum = checksum
self.name = name
self.engine = "IronPython2"
# Generate filename and filepath if source is given:
if source_dynamo_file:
filename_parts = [source_dynamo_file.basename, self.id]
self.code = node_dict_from_dyn["Code"]
# Get the name of the node:
self.name = next(
(v["Name"] for v in dynamo_file.full_dict["View"]["NodeViews"]
if v["Id"] == node_dict_from_dyn["Id"]), "")
# Generate the filename:
filename_parts = [dynamo_file.basename, self.id]
# Only add the name of the node if it's changed:
if self.name and self.name != "Python Script":
@@ -610,4 +604,17 @@ class PythonNode():
logging.debug(f"Generating filename from: {filename_parts}")
self.filename = sanitize_filename(
"_".join(filename_parts) + ".py")
self.filepath = source_dynamo_file.dirpath.joinpath(self.filename)
self.filepath = dynamo_file.dirpath.joinpath(self.filename)
elif python_file and not node_dict_from_dyn and not dynamo_file:
self.id = python_file.header_data["py_id"]
self.engine = python_file.header_data["py_engine"]
self.code = python_file.code
self.filename = python_file.basename + ".py"
self.filepath = python_file.filepath
else:
raise PythonNodeException
# Calculate checksum:
self.checksum = hashlib.md5(self.code.encode()).hexdigest()

17
tests/support.py
View File

@@ -4,6 +4,7 @@ import dyn2py
INPUT_DIR = "tests/input_files"
OUTPUT_DIR = "tests/output_files"
def cleanup_output_dir():
output_dir = pathlib.Path(OUTPUT_DIR)
if output_dir.exists():
@@ -13,9 +14,14 @@ def cleanup_output_dir():
output_dir.mkdir()
def extract_single_node_dyn():
def extract_single_node_dyn(modify_py: bool = False):
"""Extract python from single_node.dyn
File will be here: f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py"
Modified file will be here: f"{OUTPUT_DIR}/single_node_mod.py"
Args:
modify_py (bool, optional): Also do some changes on the exported file. Defaults to False.
"""
cleanup_output_dir()
@@ -23,3 +29,12 @@ def extract_single_node_dyn():
options = dyn2py.Options(python_folder=OUTPUT_DIR)
dyn = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
dyn.extract_python(options)
if modify_py:
# Open the extracted file and replace a string:
with open(f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py") as orig_py, \
open(f"{OUTPUT_DIR}/single_node_mod.py", "w") as mod_py:
for line in orig_py:
if "asd_string" in line:
line = line.replace("asd_string", "qwe_string")
mod_py.write(line)

71
tests/test_DynamoFile.py
View File

@@ -1,6 +1,8 @@
import unittest
import dyn2py
import pathlib
import shutil
import simplejson as json
from tests.support import *
@@ -8,9 +10,7 @@ from tests.support import *
class TestDynamoFile(unittest.TestCase):
# Missing methods:
# get_related_python_files
# update_python_node
# write
# update_python_node: exception
def test_init(self):
dyn2py.DynamoFile.open_files.clear()
@@ -61,8 +61,71 @@ class TestDynamoFile(unittest.TestCase):
dyn1 = dyn2py.DynamoFile(f"{INPUT_DIR}/python_nodes.dyn")
dyn2 = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
self.assertEqual(dyn1,
dyn2py.DynamoFile.get_open_file_by_uuid("3c3b4c05-9716-4e93-9360-ca0637cb5486"))
self.assertEqual(dyn2,
dyn2py.DynamoFile.get_open_file_by_uuid("76de5c79-17c5-4c74-9f90-ad99a213d339"))
def test_get_related_python_files(self):
cleanup_output_dir()
opt = dyn2py.Options(python_folder=OUTPUT_DIR)
dyn1 = dyn2py.DynamoFile(f"{INPUT_DIR}/python_nodes.dyn")
dyn2 = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
for dyn in [dyn1, dyn2]:
dyn.extract_python(options=opt)
python_files1 = dyn1.get_related_python_files(options=opt)
python_files2 = dyn2.get_related_python_files(options=opt)
self.assertEqual(len(python_files1), 6)
self.assertEqual(len(python_files2), 1)
no_python_files = dyn1.get_related_python_files()
self.assertFalse(no_python_files)
def test_write_same(self):
cleanup_output_dir()
shutil.copy(f"{INPUT_DIR}/python_nodes.dyn",
f"{OUTPUT_DIR}/python_nodes.dyn")
new_dyn = dyn2py.DynamoFile(f"{OUTPUT_DIR}/python_nodes.dyn")
new_dyn.modified = True
new_dyn.write()
with open(f"{INPUT_DIR}/python_nodes.dyn", "r", encoding="utf-8") as file1,\
open(f"{OUTPUT_DIR}/python_nodes.dyn", "r", encoding="utf-8") as file2:
json1 = json.load(file1, use_decimal=True)
json2 = json.load(file2, use_decimal=True)
self.assertEqual(json1, json2)
def test_update_and_write(self):
cleanup_output_dir()
extract_single_node_dyn(modify_py=True)
shutil.copy(f"{INPUT_DIR}/single_node.dyn",
f"{OUTPUT_DIR}/single_node.dyn")
py = dyn2py.PythonFile(f"{OUTPUT_DIR}/single_node_mod.py")
node = dyn2py.PythonNode(python_file=py)
dyn = dyn2py.DynamoFile(f"{OUTPUT_DIR}/single_node.dyn")
dyn.update_python_node(node)
self.assertTrue(dyn.modified)
self.assertTrue(node in dyn.python_nodes)
# Save the file:
dyn.write()
dyn2py.DynamoFile.open_files.clear()
shutil.copy(f"{OUTPUT_DIR}/single_node.dyn",
f"{OUTPUT_DIR}/single_node2.dyn")
dyn2 = dyn2py.DynamoFile(f"{OUTPUT_DIR}/single_node2.dyn")
node2 = dyn2.get_python_node_by_id(node_id=node.id)
self.assertTrue(node2)
self.assertEqual(node.checksum, node2.checksum)

26
tests/test_PythonNode.py
View File

@@ -1,11 +1,14 @@
import unittest
import dyn2py
import hashlib
from tests.support import *
class TestPythonNode(unittest.TestCase):
# Test needed:
# init exception
def test_init_from_dyn(self):
dyn = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
node_dict = next((n for n in dyn.full_dict["Nodes"]
@@ -18,8 +21,7 @@ class TestPythonNode(unittest.TestCase):
node = dyn2py.PythonNode(
node_dict_from_dyn=node_dict,
full_nodeviews_dict_from_dyn=node_views,
source_dynamo_file=dyn
dynamo_file=dyn
)
self.assertEqual(node.id, "1c5d99792882409e97e132b3e9f814b0")
@@ -31,26 +33,12 @@ class TestPythonNode(unittest.TestCase):
def test_init_from_py(self):
extract_single_node_dyn()
# Open the extracted file and replace a string:
with open(f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py") as orig_py, \
open(f"{OUTPUT_DIR}/single_node_mod.py", "w") as mod_py:
for line in orig_py:
if "asd_string" in line:
line = line.replace("asd_string", "qwe_string")
mod_py.write(line)
extract_single_node_dyn(modify_py=True)
py = dyn2py.PythonFile(f"{OUTPUT_DIR}/single_node_mod.py")
node = dyn2py.PythonNode(
node_id=py.header_data["py_id"],
engine=py.header_data["py_engine"],
code=py.code,
checksum=hashlib.md5(py.code.encode()).hexdigest()
)
node = dyn2py.PythonNode(python_file=py)
self.assertEqual(node.id, "1c5d99792882409e97e132b3e9f814b0")
self.assertEqual(node.engine, "CPython3")
self.assertEqual(node.checksum, "8d9091d24788a6fdfa5e1e109298b50e")