1  """
 2  Application module tests
 3  """
 4  
 5  import unittest
 6  import types
 7  
 8  from txtai.app import Application
 9  from txtai.pipeline import Pipeline
10  
11  
12  class TestApp(unittest.TestCase):
13      """
14      Application tests.
15      """
16  
17      def testConfig(self):
18          """
19          Test a file not found config exception
20          """
21  
22          with self.assertRaises(FileNotFoundError):
23              Application.read("No file here")
24  
25      def testParameter(self):
26          """
27          Test resolving application parameter
28          """
29  
30          app = Application(
31              """
32              testapp.TestPipeline:
33                  application:
34          """
35          )
36  
37          # Check that application instance is not None
38          self.assertIsNotNone(app.pipelines["testapp.TestPipeline"].application)
39  
40      def testStream(self):
41          """
42          Test workflow streams
43          """
44  
45          app = Application(
46              """
47              workflow:
48                  stream:
49                      stream:
50                          action: testapp.TestStream
51                      tasks:
52                          - nop
53                  batchstream:
54                      stream:
55                          action: testapp.TestStream
56                          batch: True
57                      tasks:
58                          - nop
59          """
60          )
61  
62          def generator():
63              yield 10
64  
65          # Test single stream
66          self.assertEqual(list(app.workflow("stream", [10])), list(range(10)))
67  
68          # Test batch stream
69          self.assertEqual(list(app.workflow("batchstream", generator())), list(range(10)))
70  
71  
72  class TestPipeline(Pipeline):
73      """
74      Test pipeline with an application parameter.
75      """
76  
77      def __init__(self, application):
78          self.application = application
79  
80  
81  class TestStream:
82      """
83      Test workflow stream
84      """
85  
86      def __call__(self, arg):
87          if isinstance(arg, types.GeneratorType):
88              for x in arg:
89                  yield from range(int(x))
90          else:
91              yield from range(int(arg))