Test the data insertion to the events_data collection
(self)
| 134 | ) |
| 135 | |
| 136 | def test_insert_events_data(self): |
| 137 | """ |
| 138 | Test the data insertion to the events_data collection |
| 139 | """ |
| 140 | event_data = EventData( |
| 141 | ip="55.66.77.88", |
| 142 | module_name="ics/veeder_root_guardian_ast", |
| 143 | date=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
| 144 | data={"content": "Test Data"} |
| 145 | ) |
| 146 | |
| 147 | insert_to_events_data_collection(event_data) |
| 148 | # wait for insert |
| 149 | time.sleep(1) |
| 150 | |
| 151 | # Find the records in the DB |
| 152 | event_records = connector.elasticsearch_events.search( |
| 153 | index='data_events', |
| 154 | body=filter_by_fields('55.66.77.88', ['ip_src']) |
| 155 | )['hits']['hits'] |
| 156 | |
| 157 | self.assertGreater(len(event_records), 0) |
| 158 | |
| 159 | event_record_data = event_records[0]['_source'] |
| 160 | # Compare the record found in the DB with the one pushed |
| 161 | self.assertEqual(event_record_data["ip_src"], event_data.ip_src) |
| 162 | self.assertEqual( |
| 163 | event_record_data["data"], |
| 164 | event_data.data |
| 165 | ) |
| 166 | connector.elasticsearch_events.delete( |
| 167 | index='data_events', |
| 168 | id=event_records[0]["_id"] |
| 169 | ) |
| 170 | |
| 171 | |
| 172 | if __name__ == '__main__': |
nothing calls this directly
no test coverage detected