(self)
| 55 | return data_sql |
| 56 | |
| 57 | def runTest(self): |
| 58 | db_con = database_utils.connect_database(self, |
| 59 | self.sgid, |
| 60 | self.sid, |
| 61 | self.did) |
| 62 | |
| 63 | if not db_con["info"] == "Database connected.": |
| 64 | raise Exception("Could not connect to database to add the schema.") |
| 65 | |
| 66 | trans_id = secrets.choice(range(1, 9999999)) |
| 67 | url = '/erd/sql/{trans_id}/{sgid}/{sid}/{did}'.format( |
| 68 | trans_id=trans_id, sgid=self.sgid, sid=self.sid, did=self.did) |
| 69 | |
| 70 | curr_dir = path.dirname(__file__) |
| 71 | |
| 72 | data_json = None |
| 73 | with open(path.join(curr_dir, 'test_sql_input_data.json')) as fp: |
| 74 | data_json = fp.read() |
| 75 | |
| 76 | response = self.tester.post(url, |
| 77 | data=data_json, |
| 78 | content_type='html/json') |
| 79 | self.assertEqual(response.status_code, 200) |
| 80 | |
| 81 | data_sql = self.get_expected_sql() |
| 82 | |
| 83 | resp_sql = json.loads(response.data.decode('utf-8'))['data'] |
| 84 | self.assertEqual(resp_sql, data_sql) |
| 85 | |
| 86 | def tearDown(self): |
| 87 | connection = utils.get_db_connection(self.server['db'], |
nothing calls this directly
no test coverage detected