indexify/python-sdk/tests/test_graph_validation.py at main · 17dev/indexify · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import unittest
from typing import List, Union

from pydantic import BaseModel

from indexify.functions_sdk.graph import Graph
from indexify.functions_sdk.indexify_functions import (
    indexify_function,
    indexify_router,
)


class TestValidations(unittest.TestCase):
    def test_function_signature_types(self):
        class ComplexType(BaseModel):
            pass

        @indexify_function()
        def node1(a: int, b: ComplexType) -> int:
            pass

        @indexify_function()
        def node2(b):
            pass

        g = Graph(
            "test-graph",
            start_node=node1,
        )

        msg = "Input param b in node2 has empty type annotation"

        with self.assertRaises(Exception) as cm:
            g.add_edge(node1, node2)

        self.assertEqual(msg, str(cm.exception))

    def test_function_return_type_annotation(self):
        class ComplexType(BaseModel):
            pass

        @indexify_function()
        def node1(a: int, b: ComplexType) -> int:
            pass

        @indexify_function()
        def node2(b: float):
            pass

        g = Graph(
            "test-graph",
            start_node=node1,
        )

        msg = "Function node2 has empty return type annotation"

        with self.assertRaises(Exception) as cm:
            g.add_edge(node1, node2)

        self.assertEqual(msg, str(cm.exception))

    def test_callables_are_in_added_nodes(self):
        class ComplexType(BaseModel):
            pass

        def node1(a: int, b: ComplexType) -> int:
            pass

        @indexify_function()
        def node2(b: int) -> ComplexType:
            pass

        with self.assertRaises(Exception) as cm:
            g = Graph(
                "test-graph",
                start_node=node1,
            )

            g.add_edge(node1, node2)

        msg = "Unable to add node of type `<class 'function'>`. Required, `IndexifyFunction` or `IndexifyRouter`"
        self.assertEqual(msg, str(cm.exception))

    def test_router_callables_are_in_added_nodes_union(self):
        @indexify_function()
        def node0(a: int) -> int:
            pass

        @indexify_function()
        def node1(a: int) -> int:
            pass

        @indexify_function()
        def node2(a: int) -> int:
            pass

        @indexify_function()
        def node3(a: int) -> int:
            pass

        @indexify_router()
        def router(a: int) -> List[Union[node1, node3]]:
            pass

        @indexify_router()
        def router2(a: int) -> Union[node1, node3]:
            pass

        with self.assertRaises(Exception) as cm:
            g = Graph(
                "test-graph",
                start_node=node0,
            )

            g.add_edge(node0, router)
            g.route(router, [node1, node2])
        msg = "Unable to find node3 in to_nodes ['node1', 'node2']"
        self.assertEqual(msg, str(cm.exception))

        with self.assertRaises(Exception) as cm:
            g = Graph(
                "test-graph",
                start_node=node0,
            )

            g.add_edge(node0, router)
            g.route(router2, [node1, node2])
        msg = "Unable to find node3 in to_nodes ['node1', 'node2']"
        self.assertEqual(msg, str(cm.exception))

    def test_route_validation_with_valid_return_type_signature(self):
        @indexify_function()
        def start() -> int:
            return 1

        @indexify_function()
        def end() -> int:
            return 1

        @indexify_router()
        def route1(**kwargs: dict) -> Union[int, float]:
            return 10

        g = Graph(name="test", start_node=start)
        g.add_edge(start, route1)
        g.route(route1, [end, end])


if __name__ == "__main__":
    unittest.main()