| Total Complexity | 2 |
| Total Lines | 36 |
| Duplicated Lines | 0 % |
| Changes | 0 | ||
| 1 | import integ_test_base |
||
| 2 | |||
| 3 | |||
| 4 | class TestCustomEvaluateTimeout(integ_test_base.IntegTestBase): |
||
| 5 | def _get_evaluate_timeout(self) -> str: |
||
| 6 | return '5' |
||
| 7 | |||
| 8 | def test_custom_evaluate_timeout_with_script(self): |
||
| 9 | payload = ( |
||
| 10 | ''' |
||
| 11 | { |
||
| 12 | "data": { "_arg1": 1 }, |
||
| 13 | "script": |
||
| 14 | "import time\\nwhile True:\\n time.sleep(1)\\nreturn 1" |
||
| 15 | } |
||
| 16 | ''') |
||
| 17 | headers = { |
||
| 18 | 'Content-Type': |
||
| 19 | "application/json", |
||
| 20 | 'TabPy-Client': |
||
| 21 | "Integration test for testing custom evaluate timeouts with " |
||
| 22 | "scripts." |
||
| 23 | } |
||
| 24 | |||
| 25 | conn = self._get_connection() |
||
| 26 | conn.request('POST', '/evaluate', payload, headers) |
||
| 27 | res = conn.getresponse() |
||
| 28 | actual_error_message = res.read().decode('utf-8') |
||
| 29 | |||
| 30 | self.assertEqual( |
||
| 31 | '{"message": ' |
||
| 32 | '"User defined script timed out. Timeout is set to 5.0 s.", ' |
||
| 33 | '"info": {}}', |
||
| 34 | actual_error_message) |
||
| 35 | self.assertEqual(408, res.status) |
||
| 36 |