| 113 | assert len(fuzzy3) == 2 |
| 114 | |
| 115 | async def test_fact_origin(self, knowledge_svc, ability, executor): |
| 116 | texecutor = executor(name='sh', platform='darwin', command='mkdir test', cleanup='rm -rf test') |
| 117 | tability = ability(ability_id='123', executors=[texecutor], repeatable=True, buckets=['test']) |
| 118 | link = Link.load(dict(command='', paw='n1234', ability=tability, executor=next(tability.executors), status=0, |
| 119 | id='ganymede')) |
| 120 | type1_fact = Fact(trait='t1', value='d1', score=1, collected_by=['thin_air'], technique_id='T1234', |
| 121 | links=[link.id], origin_type=OriginType.LEARNED) |
| 122 | type2_fact = Fact(trait='t2', value='d2', score=1, collected_by=['thin_air'], technique_id='T1234', |
| 123 | links=[link.id], origin_type=OriginType.LEARNED) |
| 124 | type3_fact = Fact(trait='t3', value='d3', score=1, collected_by=['tiny_lightning_bolts_running_through_sand'], |
| 125 | technique_id='T1234', origin_type=OriginType.SEEDED, source="Europa") |
| 126 | await knowledge_svc.add_fact(type1_fact) |
| 127 | await knowledge_svc.add_fact(type2_fact) |
| 128 | await knowledge_svc.add_fact(type3_fact) |
| 129 | origin_1, type_1 = await knowledge_svc.get_fact_origin(type1_fact) |
| 130 | origin_2, type_2 = await knowledge_svc.get_fact_origin(type2_fact.trait) |
| 131 | origin_3, type_3 = await knowledge_svc.get_fact_origin(type3_fact.trait) |
| 132 | assert origin_1 == link.id |
| 133 | assert origin_2 == link.id |
| 134 | assert origin_3 == 'Europa' |
| 135 | assert type_1 == OriginType.LEARNED |
| 136 | assert type_2 == OriginType.LEARNED |
| 137 | assert type_3 == OriginType.SEEDED |