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

Browse Source
This commit is contained in:
Péter Gyetvai
2023-03-13 23:37:46 +01:00
parent 7ef6a0f886
commit 02dea55518
6 changed files with 158 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
TODO.md
View File

@@ -4,9 +4,9 @@
- [ ] Commandline
- [x] File
- [ ] DynamoFile
- [ ] PythonFile
- [ ] PythonNode
- [x] DynamoFile
- [x] PythonFile
- [x] PythonNode
- [ ] run()
## CI/CD

34
dyn2py/files.py
View File

@@ -4,6 +4,7 @@ import hashlib
import pathlib
import textwrap
import logging
import os
from datetime import datetime
from decimal import Decimal
from pathvalidate import sanitize_filename
@@ -362,7 +363,7 @@ class PythonFile(File):
header_data: dict
"""Parsed dict from the header of a python file."""
text: str
"""Full contents of the file before writing."""
"""Full contents of the file."""
open_files: set["PythonFile"] = set()
"""A set of open Python files."""
@@ -421,6 +422,7 @@ class PythonFile(File):
python_node.code
])
self.code = python_node.code
self.modified = True
else:
@@ -449,6 +451,7 @@ class PythonFile(File):
with open(self.filepath, mode="r", newline="", encoding="utf-8") as input_py:
python_lines = input_py.readlines()
self.text = os.linesep.join(python_lines)
self.header_data = {}
header_separator_count = 0
code_start_line = 0
@@ -490,13 +493,7 @@ class PythonFile(File):
if not options:
options = Options()
# Check if it was already opened:
dynamo_file = DynamoFile.get_open_file_by_uuid(
self.header_data["dyn_uuid"])
# Open if it's the first time:
if not dynamo_file:
dynamo_file = self.get_source_dynamo_file()
dynamo_file = self.get_source_dynamo_file()
new_python_node = PythonNode(python_file=self)
@@ -519,22 +516,25 @@ class PythonFile(File):
"""Get the source Dynamo file of this PythonFile
Raises:
FileNotFoundError: The dynamo file not found
DynamoFileException: The uuid of the dynamo file changed
Returns:
DynamoFile: The DynamoFile
"""
dynamo_file = DynamoFile(
pathlib.Path(self.header_data["dyn_path"]))
if not dynamo_file.exists:
raise FileNotFoundError(
f"Dynamo graph not found: {dynamo_file.filepath}")
# Check if it was already opened:
dynamo_file = DynamoFile.get_open_file_by_uuid(
self.header_data["dyn_uuid"])
# Check if uuid is ok:
if not dynamo_file.uuid == self.header_data["dyn_uuid"]:
raise DynamoFileException(f"Dynamo graph uuid changed!")
# Open if it's the first time:
if not dynamo_file:
dynamo_file = DynamoFile(
pathlib.Path(self.header_data["dyn_path"]))
# Check if uuid is ok:
if not dynamo_file.uuid == self.header_data["dyn_uuid"]:
raise DynamoFileException(f"Dynamo graph uuid changed!")
return dynamo_file

33
tests/test_DynamoFile.py
View File

@@ -3,15 +3,13 @@ import dyn2py
import pathlib
import shutil
import simplejson as json
from dyn2py.files import DynamoFile
from tests.support import *
class TestDynamoFile(unittest.TestCase):
# Missing methods:
# update_python_node: exception
def test_init(self):
dyn2py.DynamoFile.open_files.clear()
@@ -19,7 +17,7 @@ class TestDynamoFile(unittest.TestCase):
self.assertEqual(dyn.uuid, "3c3b4c05-9716-4e93-9360-ca0637cb5486")
self.assertEqual(dyn.name, "python_nodes")
self.assertTrue(dyn in dyn.open_files)
self.assertTrue(dyn in DynamoFile.open_files)
# Dynamo 1 file:
with self.assertRaises(dyn2py.DynamoFileException):
@@ -39,7 +37,7 @@ class TestDynamoFile(unittest.TestCase):
self.assertEqual(len(dyn.python_nodes), 6)
self.assertTrue(py_node)
self.assertTrue(py_node in dyn.python_nodes)
self.assertIn(py_node, dyn.python_nodes)
self.assertEqual(py_node.checksum, "1f3d9e6153804fe1ed37571a9cda8e26")
with self.assertRaises(dyn2py.PythonNodeNotFoundException):
@@ -103,29 +101,36 @@ class TestDynamoFile(unittest.TestCase):
self.assertEqual(json1, json2)
def test_update_and_write(self):
cleanup_output_dir()
extract_single_node_dyn(modify_py=True)
# Create a copy to update it:
shutil.copy(f"{INPUT_DIR}/single_node.dyn",
f"{OUTPUT_DIR}/single_node.dyn")
# Read back the modified py, and update:
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)
node1 = dyn2py.PythonNode(python_file=py)
dyn1 = dyn2py.DynamoFile(f"{OUTPUT_DIR}/single_node.dyn")
dyn1.update_python_node(node1)
self.assertTrue(dyn.modified)
self.assertTrue(node in dyn.python_nodes)
self.assertTrue(dyn1.modified)
self.assertIn(node1, dyn1.python_nodes)
# Save the file:
dyn.write()
dyn1.write()
dyn2py.DynamoFile.open_files.clear()
shutil.copy(f"{OUTPUT_DIR}/single_node.dyn",
f"{OUTPUT_DIR}/single_node2.dyn")
# Check if the node in the copy:
dyn2 = dyn2py.DynamoFile(f"{OUTPUT_DIR}/single_node2.dyn")
node2 = dyn2.get_python_node_by_id(node_id=node.id)
node2 = dyn2.get_python_node_by_id(node_id=node1.id)
self.assertTrue(node2)
self.assertEqual(node.checksum, node2.checksum)
self.assertEqual(node1.checksum, node2.checksum)
with self.assertRaises(dyn2py.PythonNodeNotFoundException):
node2.id = "wrong_id"
dyn2.update_python_node(node2)

2
tests/test_File.py
View File

@@ -35,7 +35,7 @@ class TestFile(unittest.TestCase):
self.assertEqual(the_file.extension, ".dyn")
self.assertFalse(the_file.modified)
self.assertEqual(the_file.__class__, dyn2py.DynamoFile)
self.assertIs(type(the_file), dyn2py.DynamoFile)
def test_init_newfile(self):
paths = [

111
tests/test_PythonFile.py
View File

@@ -1,19 +1,106 @@
import unittest
import dyn2py
import shutil
from time import sleep
from tests.support import *
class TestPythonFile(unittest.TestCase):
pass
# Missing variables:
# code
# header_data
# text
# open_files
def test_init(self):
extract_single_node_dyn()
# Missing methods:
# generate_text()
# update_dynamo()
# get_source_dynamo_file()
# read()
# write()
py1 = dyn2py.PythonFile(
f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
dyn2py.DynamoFile.open_files.clear()
dyn = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
node = list(dyn.python_nodes)[0]
py2 = dyn2py.PythonFile(filepath=node.filepath,
dynamo_file=dyn, python_node=node)
for py in [py1, py2]:
self.assertEqual(len(py.code.split("\n")), 17)
self.assertEqual(len(py.text.split("\r\n")), 31)
self.assertIs(type(py.header_data), dict)
self.assertTrue(py in dyn2py.PythonFile.open_files)
def test_update_dynamo(self):
extract_single_node_dyn(modify_py=True)
dyn2py.DynamoFile.open_files.clear()
dyn2py.PythonFile.open_files.clear()
py1 = dyn2py.PythonFile(
f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
dyn1 = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
# This shouldn't modify, as this file was not updated:
py1.update_dynamo()
self.assertFalse(dyn1.modified)
# The modified files should update:
py2 = dyn2py.PythonFile(f"{OUTPUT_DIR}/single_node_mod.py")
py2.update_dynamo()
self.assertTrue(dyn1.modified)
dyn2py.DynamoFile.open_files.clear()
# Make sure that the dyn file is newer:
sleep(1)
shutil.copy(f"{INPUT_DIR}/single_node.dyn",
f"{OUTPUT_DIR}/single_node.dyn")
dyn2 = dyn2py.DynamoFile(f"{OUTPUT_DIR}/single_node.dyn")
py2.header_data["dyn_path"] = f"{OUTPUT_DIR}/single_node.dyn"
self.assertFalse(py2.is_newer(dyn2))
# This shouldn't modify, as it is newer:
py2.update_dynamo()
self.assertFalse(dyn2.modified)
# It should work with force:
opt = dyn2py.Options(force=True)
py2.update_dynamo(options=opt)
self.assertTrue(dyn2.modified)
def test_get_source_dynamo_file(self):
extract_single_node_dyn()
dyn2py.DynamoFile.open_files.clear()
dyn2py.PythonFile.open_files.clear()
py1 = dyn2py.PythonFile(
f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
dyn1 = py1.get_source_dynamo_file()
self.assertEqual(len(dyn2py.DynamoFile.open_files), 1)
self.assertIn(dyn1, dyn2py.DynamoFile.open_files)
dyn2 = py1.get_source_dynamo_file()
self.assertIs(dyn1, dyn2)
dyn2py.DynamoFile.open_files.clear()
with self.assertRaises(dyn2py.DynamoFileException):
py1.header_data["dyn_uuid"] = "wrong-uuid"
py1.get_source_dynamo_file()
def test_write(self):
extract_single_node_dyn()
dyn2py.DynamoFile.open_files.clear()
dyn2py.PythonFile.open_files.clear()
py1 = dyn2py.PythonFile(
f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
dyn1 = py1.get_source_dynamo_file()
node = list(dyn1.python_nodes)[0]
py2 = dyn2py.PythonFile(
node.filepath, dynamo_file=dyn1, python_node=node)
self.assertIsNot(py1, py2)
self.assertEqual(py1.code, py2.code)
for d in py1.header_data:
if not d == "dyn2py_extracted":
self.assertEqual(py1.header_data[d], py2.header_data[d])

24
tests/test_PythonNode.py
View File

@@ -6,9 +6,6 @@ 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"]
@@ -17,8 +14,6 @@ class TestPythonNode(unittest.TestCase):
# Found a node:
self.assertTrue(node_dict)
node_views = dyn.full_dict["View"]["NodeViews"]
node = dyn2py.PythonNode(
node_dict_from_dyn=node_dict,
dynamo_file=dyn
@@ -42,3 +37,22 @@ class TestPythonNode(unittest.TestCase):
self.assertEqual(node.id, "1c5d99792882409e97e132b3e9f814b0")
self.assertEqual(node.engine, "CPython3")
self.assertEqual(node.checksum, "8d9091d24788a6fdfa5e1e109298b50e")
def test_init_exception(self):
dyn = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
node_dict = next((n for n in dyn.full_dict["Nodes"]
if n["NodeType"] == "PythonScriptNode"), {})
extract_single_node_dyn(modify_py=True)
py = dyn2py.PythonFile(f"{OUTPUT_DIR}/single_node_mod.py")
with self.assertRaises(dyn2py.PythonNodeException):
node1 = dyn2py.PythonNode(
node_dict_from_dyn=node_dict,
dynamo_file=dyn,
python_file=py
)
node2 = dyn2py.PythonNode()