External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-10 02:47:36 -02:30
Code Issues Packages Projects Releases Wiki Activity

added fact view tests

Browse Source 
* Add related link from host to fact_versions
* Add related link from fact_versions to fact_view
This commit is contained in:
Chris Meyers
2015-05-08 21:13:42 -04:00
parent 0c6a4782fd
commit 5ff696f708
17 changed files with 547 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
awx/fact/tests/__init__.py
View File

@@ -5,3 +5,4 @@ from __future__ import absolute_import
from .models import * # noqa
from .utils import * # noqa
from .base import * # noqa

202
awx/fact/tests/base.py Normal file
View File

@@ -0,0 +1,202 @@
# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved
# Python
from __future__ import absolute_import
from copy import deepcopy
from datetime import datetime
from django.utils.timezone import now
# Django
from django.conf import settings
import django
# MongoEngine
from mongoengine.connection import get_db, ConnectionError
# AWX
from awx.fact.models.fact import * # noqa
TEST_FACT_ANSIBLE = {
"ansible_swapfree_mb" : 4092,
"ansible_default_ipv6" : {
},
"ansible_distribution_release" : "trusty",
"ansible_system_vendor" : "innotek GmbH",
"ansible_os_family" : "Debian",
"ansible_all_ipv4_addresses" : [
"192.168.1.145"
],
"ansible_lsb" : {
"release" : "14.04",
"major_release" : "14",
"codename" : "trusty",
"id" : "Ubuntu",
"description" : "Ubuntu 14.04.2 LTS"
},
}
TEST_FACT_PACKAGES = [
{
"name": "accountsservice",
"architecture": "amd64",
"source": "apt",
"version": "0.6.35-0ubuntu7.1"
},
{
"name": "acpid",
"architecture": "amd64",
"source": "apt",
"version": "1:2.0.21-1ubuntu2"
},
{
"name": "adduser",
"architecture": "all",
"source": "apt",
"version": "3.113+nmu3ubuntu3"
},
]
TEST_FACT_SERVICES = [
{
"source" : "upstart",
"state" : "waiting",
"name" : "ureadahead-other",
"goal" : "stop"
},
{
"source" : "upstart",
"state" : "running",
"name" : "apport",
"goal" : "start"
},
{
"source" : "upstart",
"state" : "waiting",
"name" : "console-setup",
"goal" : "stop"
},
]
class MongoDBRequired(django.test.TestCase):
def setUp(self):
# Drop mongo database
try:
self.db = get_db()
self.db.connection.drop_database(settings.MONGO_DB)
except ConnectionError:
self.skipTest('MongoDB connection failed')
class BaseFactTestMixin(MongoDBRequired):
pass
class BaseFactTest(BaseFactTestMixin, MongoDBRequired):
pass
class FactScanBuilder(object):
def __init__(self):
self.facts_data = {}
self.hostname_data = []
self.host_objs = []
self.fact_objs = []
self.version_objs = []
self.timestamps = []
def add_fact(self, module, facts):
self.facts_data[module] = facts
def add_hostname(self, hostname):
self.hostname_data.append(hostname)
def build(self, scan_count, host_count):
if len(self.facts_data) == 0:
raise RuntimeError("No fact data to build populate scans. call add_fact()")
if (len(self.hostname_data) > 0 and len(self.hostname_data) != host_count):
raise RuntimeError("Registered number of hostnames %d does not match host_count %d" % (len(self.hostname_data), host_count))
if len(self.hostname_data) == 0:
self.hostname_data = ['hostname_%s' % i for i in range(0, host_count)]
self.host_objs = [FactHost(hostname=hostname).save() for hostname in self.hostname_data]
for i in range(0, scan_count):
scan = {}
scan_version = {}
timestamp = now().replace(year=2015 - i, microsecond=0)
for module in self.facts_data:
fact_objs = []
version_objs = []
for host in self.host_objs:
(fact_obj, version_obj) = Fact.add_fact(timestamp=timestamp,
host=host,
module=module,
fact=self.facts_data[module])
fact_objs.append(fact_obj)
version_objs.append(version_obj)
scan[module] = fact_objs
scan_version[module] = version_objs
self.fact_objs.append(scan)
self.version_objs.append(scan_version)
self.timestamps.append(timestamp)
def get_scan(self, index, module=None):
res = None
res = self.fact_objs[index]
if module:
res = res[module]
return res
def get_scans(self, index_start=None, index_end=None):
if index_start is None:
index_start = 0
if index_end is None:
index_end = len(self.fact_objs)
return self.fact_objs[index_start:index_end]
def get_scan_version(self, index, module=None):
res = None
res = self.version_objs[index]
if module:
res = res[module]
return res
def get_scan_versions(self, index_start=None, index_end=None):
if index_start is None:
index_start = 0
if index_end is None:
index_end = len(self.version_objs)
return self.version_objs[index_start:index_end]
def get_hostname(self, index):
return self.host_objs[index].hostname
def get_hostnames(self, index_start=None, index_end=None):
if index_start is None:
index_start = 0
if index_end is None:
index_end = len(self.host_objs)
return [self.host_objs[i].hostname for i in range(index_start, index_end)]
def get_scan_count(self):
return len(self.fact_objs)
def get_host_count(self):
return len(self.host_objs)
def get_timestamp(self, index):
return self.timestamps[index]
def get_timestamps(self, index_start=None, index_end=None):
if not index_start:
index_start = 0
if not index_end:
len(self.timestamps)
return self.timestamps[index_start:index_end]

57
awx/fact/tests/models/fact/base.py
View File

@@ -1,57 +0,0 @@
# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved
# Python
from __future__ import absolute_import
from awx.main.tests.base import BaseTest, MongoDBRequired
from copy import deepcopy
from datetime import datetime
# AWX
from awx.fact.models.fact import * # noqa
'''
Helper functions (i.e. create_host_document) expect the structure:
{
'hostname': 'hostname1',
'add_fact_data': {
'timestamp': datetime.now(),
'host': None,
'module': 'packages',
'fact': ...
}
}
'''
class BaseFactTest(BaseTest, MongoDBRequired):
@staticmethod
def _normalize_timestamp(timestamp):
return timestamp.replace(microsecond=0)
@staticmethod
def normalize_timestamp(data):
data['add_fact_data']['timestamp'] = BaseFactTest._normalize_timestamp(data['add_fact_data']['timestamp'])
def create_host_document(self, data):
data['add_fact_data']['host'] = FactHost(hostname=data['hostname']).save()
def create_fact_scans(self, data, host_count=1, scan_count=1):
timestamps = []
self.fact_data = []
self.fact_objs = []
self.hostnames = [FactHost(hostname='%s_%s' % (data['hostname'], i)).save() for i in range(0, host_count)]
for i in range(0, scan_count):
self.fact_data.append([])
self.fact_objs.append([])
for j in range(0, host_count):
data = deepcopy(data)
t = datetime.now().replace(year=2015 - i, microsecond=0)
data['add_fact_data']['timestamp'] = t
data['add_fact_data']['host'] = self.hostnames[j]
(f, v) = Fact.add_fact(**data['add_fact_data'])
timestamps.append(t)
self.fact_data[i].append(data)
self.fact_objs[i].append(f)
return timestamps

93
awx/fact/tests/models/fact/fact_get_single_facts.py
View File

@@ -9,61 +9,17 @@ from datetime import datetime
# AWX
from awx.fact.models.fact import * # noqa
from .base import BaseFactTest
from awx.fact.tests.base import BaseFactTest, FactScanBuilder, TEST_FACT_PACKAGES
__all__ = ['FactGetSingleFactsTest', 'FactGetSingleFactsMultipleScansTest',]
TEST_FACT_PACKAGES = [
{
"name": "accountsservice",
"architecture": "amd64",
"source": "apt",
"version": "0.6.35-0ubuntu7.1"
},
{
"name": "acpid",
"architecture": "amd64",
"source": "apt",
"version": "1:2.0.21-1ubuntu2"
},
{
"name": "adduser",
"architecture": "all",
"source": "apt",
"version": "3.113+nmu3ubuntu3"
},
]
TEST_FACT_DATA = {
'hostname': 'hostname_%d',
'add_fact_data': {
'timestamp': datetime.now(),
'host': None,
'module': 'packages',
'fact': TEST_FACT_PACKAGES,
}
}
TEST_FACT_NESTED_DATA = {
'hostname': 'hostname_%d',
'add_fact_data': {
'timestamp': datetime.now(),
'host': None,
'module': 'packages',
'fact': {
'nested': TEST_FACT_PACKAGES
},
}
}
class FactGetSingleFactsTest(BaseFactTest):
def setUp(self):
super(FactGetSingleFactsTest, self).setUp()
self.host_count = 20
self.timestamp = datetime.now().replace(year=2016)
self.create_fact_scans(TEST_FACT_DATA, self.host_count, scan_count=1)
self.hosts = [self.hostnames[i].hostname for i in range(0, self.host_count)]
self.builder = FactScanBuilder()
self.builder.add_fact('packages', TEST_FACT_PACKAGES)
self.builder.add_fact('nested', TEST_FACT_PACKAGES)
self.builder.build(scan_count=1, host_count=20)
def check_query_results(self, facts_known, facts):
self.assertIsNotNone(facts)
@@ -95,50 +51,47 @@ class FactGetSingleFactsTest(BaseFactTest):
self.assertEqual(fact.fact['nested'][0]['name'], 'acpid')
def test_single_host(self):
self.hosts = [self.hostnames[i].hostname for i in range(0, 1)]
facts = Fact.get_single_facts(self.hosts, 'name', 'acpid', self.timestamp, 'packages')
facts = Fact.get_single_facts(self.builder.get_hostnames(0, 1), 'name', 'acpid', self.builder.get_timestamp(0), 'packages')
self.check_query_results(self.fact_objs[0][:1], facts)
self.check_query_results(self.builder.get_scan(0, 'packages')[:1], facts)
def test_all(self):
facts = Fact.get_single_facts(self.hosts, 'name', 'acpid', self.timestamp, 'packages')
facts = Fact.get_single_facts(self.builder.get_hostnames(), 'name', 'acpid', self.builder.get_timestamp(0), 'packages')
self.check_query_results(self.fact_objs[0], facts)
self.check_query_results(self.builder.get_scan(0, 'packages'), facts)
def test_subset_hosts(self):
self.hosts = [self.hostnames[i].hostname for i in range(0, (self.host_count / 2))]
facts = Fact.get_single_facts(self.hosts, 'name', 'acpid', self.timestamp, 'packages')
host_count = (self.builder.get_host_count() / 2)
facts = Fact.get_single_facts(self.builder.get_hostnames(0, host_count), 'name', 'acpid', self.builder.get_timestamp(0), 'packages')
self.check_query_results(self.fact_objs[0][:(self.host_count / 2)], facts)
self.check_query_results(self.builder.get_scan(0, 'packages')[:host_count], facts)
def test_get_single_facts_nested(self):
facts = Fact.get_single_facts(self.hosts, 'nested.name', 'acpid', self.timestamp, 'packages')
facts = Fact.get_single_facts(self.builder.get_hostnames(), 'nested.name', 'acpid', self.builder.get_timestamp(0), 'packages')
self.check_query_results_nested(facts)
class FactGetSingleFactsMultipleScansTest(BaseFactTest):
def setUp(self):
super(FactGetSingleFactsMultipleScansTest, self).setUp()
self.create_fact_scans(TEST_FACT_DATA, host_count=10, scan_count=10)
self.builder = FactScanBuilder()
self.builder.add_fact('packages', TEST_FACT_PACKAGES)
self.builder.build(scan_count=10, host_count=10)
def test_1_host(self):
timestamp = datetime.now().replace(year=2016)
facts = Fact.get_single_facts([self.hostnames[0].hostname], 'name', 'acpid', timestamp, 'packages')
facts = Fact.get_single_facts(self.builder.get_hostnames(0, 1), 'name', 'acpid', self.builder.get_timestamp(0), 'packages')
self.assertEqual(len(facts), 1)
self.assertEqual(facts[0], self.fact_objs[0][0])
self.assertEqual(facts[0], self.builder.get_scan(0, 'packages')[0])
def test_multiple_hosts(self):
timestamp = datetime.now().replace(year=2016)
hosts = [self.hostnames[i].hostname for i in range(0, 3)]
facts = Fact.get_single_facts(hosts, 'name', 'acpid', timestamp, 'packages')
facts = Fact.get_single_facts(self.builder.get_hostnames(0, 3), 'name', 'acpid', self.builder.get_timestamp(0), 'packages')
self.assertEqual(len(facts), 3)
for i, fact in enumerate(facts):
self.assertEqual(fact, self.fact_objs[0][i])
self.assertEqual(fact, self.builder.get_scan(0, 'packages')[i])
def test_middle_of_timeline(self):
timestamp = datetime.now().replace(year=2013)
hosts = [self.hostnames[i].hostname for i in range(0, 3)]
facts = Fact.get_single_facts(hosts, 'name', 'acpid', timestamp, 'packages')
facts = Fact.get_single_facts(self.builder.get_hostnames(0, 3), 'name', 'acpid', self.builder.get_timestamp(4), 'packages')
self.assertEqual(len(facts), 3)
for i, fact in enumerate(facts):
self.assertEqual(fact, self.fact_objs[2][i])
self.assertEqual(fact, self.builder.get_scan(4, 'packages')[i])

118
awx/fact/tests/models/fact/fact_simple.py
View File

@@ -3,66 +3,30 @@
# Python
from __future__ import absolute_import
from datetime import datetime
from django.utils.timezone import now
from copy import deepcopy
from dateutil.relativedelta import relativedelta
# Django
# AWX
from awx.fact.models.fact import * # noqa
from .base import BaseFactTest
from awx.fact.tests.base import BaseFactTest, FactScanBuilder, TEST_FACT_PACKAGES
__all__ = ['FactHostTest', 'FactTest', 'FactGetHostVersionTest', 'FactGetHostTimelineTest']
TEST_FACT_DATA = {
'hostname': 'hostname1',
'add_fact_data': {
'timestamp': datetime.now(),
'host': None,
'module': 'packages',
'fact': {
"accountsservice": [
{
"architecture": "amd64",
"name": "accountsservice",
"source": "apt",
"version": "0.6.35-0ubuntu7.1"
}
],
"acpid": [
{
"architecture": "amd64",
"name": "acpid",
"source": "apt",
"version": "1:2.0.21-1ubuntu2"
}
],
"adduser": [
{
"architecture": "all",
"name": "adduser",
"source": "apt",
"version": "3.113+nmu3ubuntu3"
}
],
},
}
}
# Strip off microseconds because mongo has less precision
BaseFactTest.normalize_timestamp(TEST_FACT_DATA)
class FactHostTest(BaseFactTest):
def test_create_host(self):
host = FactHost(hostname=TEST_FACT_DATA['hostname'])
host = FactHost(hostname='hosty')
host.save()
host = FactHost.objects.get(hostname=TEST_FACT_DATA['hostname'])
host = FactHost.objects.get(hostname='hosty')
self.assertIsNotNone(host, "Host added but not found")
self.assertEqual(TEST_FACT_DATA['hostname'], host.hostname, "Gotten record hostname does not match expected hostname")
self.assertEqual('hosty', host.hostname, "Gotten record hostname does not match expected hostname")
# Ensure an error is raised for .get() that doesn't match a record.
def test_get_host_id_no_result(self):
host = FactHost(hostname=TEST_FACT_DATA['hostname'])
host = FactHost(hostname='hosty')
host.save()
self.assertRaises(FactHost.DoesNotExist, FactHost.objects.get, hostname='doesnotexist')
@@ -70,70 +34,64 @@ class FactHostTest(BaseFactTest):
class FactTest(BaseFactTest):
def setUp(self):
super(FactTest, self).setUp()
self.create_host_document(TEST_FACT_DATA)
def test_add_fact(self):
(f_obj, v_obj) = Fact.add_fact(**TEST_FACT_DATA['add_fact_data'])
timestamp = now().replace(microsecond=0)
host = FactHost(hostname="hosty").save()
(f_obj, v_obj) = Fact.add_fact(host=host, timestamp=timestamp, module='packages', fact=TEST_FACT_PACKAGES)
f = Fact.objects.get(id=f_obj.id)
v = FactVersion.objects.get(id=v_obj.id)
self.assertEqual(f.id, f_obj.id)
self.assertEqual(f.module, TEST_FACT_DATA['add_fact_data']['module'])
self.assertEqual(f.fact, TEST_FACT_DATA['add_fact_data']['fact'])
self.assertEqual(f.timestamp, TEST_FACT_DATA['add_fact_data']['timestamp'])
self.assertEqual(f.module, 'packages')
self.assertEqual(f.fact, TEST_FACT_PACKAGES)
self.assertEqual(f.timestamp, timestamp)
# host relationship created
self.assertEqual(f.host.id, TEST_FACT_DATA['add_fact_data']['host'].id)
self.assertEqual(f.host.id, host.id)
# version created and related
self.assertEqual(v.id, v_obj.id)
self.assertEqual(v.timestamp, TEST_FACT_DATA['add_fact_data']['timestamp'])
self.assertEqual(v.host.id, TEST_FACT_DATA['add_fact_data']['host'].id)
self.assertEqual(v.timestamp, timestamp)
self.assertEqual(v.host.id, host.id)
self.assertEqual(v.fact.id, f_obj.id)
self.assertEqual(v.fact.module, TEST_FACT_DATA['add_fact_data']['module'])
self.assertEqual(v.fact.module, 'packages')
class FactGetHostVersionTest(BaseFactTest):
def setUp(self):
super(FactGetHostVersionTest, self).setUp()
self.create_host_document(TEST_FACT_DATA)
self.t1 = datetime.now().replace(second=1, microsecond=0)
self.t2 = datetime.now().replace(second=2, microsecond=0)
data = deepcopy(TEST_FACT_DATA)
data['add_fact_data']['timestamp'] = self.t1
(self.f1, self.v1) = Fact.add_fact(**data['add_fact_data'])
data = deepcopy(TEST_FACT_DATA)
data['add_fact_data']['timestamp'] = self.t2
(self.f2, self.v2) = Fact.add_fact(**data['add_fact_data'])
self.builder = FactScanBuilder()
self.builder.add_fact('packages', TEST_FACT_PACKAGES)
self.builder.build(scan_count=2, host_count=1)
def test_get_host_version_exact_timestamp(self):
fact = Fact.get_host_version(hostname=TEST_FACT_DATA['hostname'], timestamp=self.t1, module=TEST_FACT_DATA['add_fact_data']['module'])
self.assertIsNotNone(fact, "Set of Facts not found")
self.assertEqual(self.f1.id, fact.id)
self.assertEqual(self.f1.fact, fact.fact)
fact_known = self.builder.get_scan(0, 'packages')[0]
fact = Fact.get_host_version(hostname=self.builder.get_hostname(0), timestamp=self.builder.get_timestamp(0), module='packages')
self.assertIsNotNone(fact)
self.assertEqual(fact_known, fact)
def test_get_host_version_lte_timestamp(self):
t3 = datetime.now().replace(second=3, microsecond=0)
fact = Fact.get_host_version(hostname=TEST_FACT_DATA['hostname'], timestamp=t3, module=TEST_FACT_DATA['add_fact_data']['module'])
self.assertEqual(self.f1.id, fact.id)
self.assertEqual(self.f1.fact, fact.fact)
timestamp = self.builder.get_timestamp(0) + relativedelta(days=1)
fact_known = self.builder.get_scan(0, 'packages')[0]
fact = Fact.get_host_version(hostname=self.builder.get_hostname(0), timestamp=timestamp, module='packages')
self.assertIsNotNone(fact)
self.assertEqual(fact_known, fact)
def test_get_host_version_none(self):
t3 = deepcopy(self.t1).replace(second=0)
fact = Fact.get_host_version(hostname=TEST_FACT_DATA['hostname'], timestamp=t3, module=TEST_FACT_DATA['add_fact_data']['module'])
timestamp = self.builder.get_timestamp(0) - relativedelta(years=20)
fact = Fact.get_host_version(hostname=self.builder.get_hostname(0), timestamp=timestamp, module='packages')
self.assertIsNone(fact)
class FactGetHostTimelineTest(BaseFactTest):
def setUp(self):
super(FactGetHostTimelineTest, self).setUp()
#self.create_host_document(TEST_FACT_DATA)
self.scans = 20
self.timestamps = self.create_fact_scans(TEST_FACT_DATA, host_count=1, scan_count=self.scans)
self.builder = FactScanBuilder()
self.builder.add_fact('packages', TEST_FACT_PACKAGES)
self.builder.build(scan_count=20, host_count=1)
def test_get_host_timeline_ok(self):
timestamps = Fact.get_host_timeline(hostname=self.hostnames[0].hostname, module=TEST_FACT_DATA['add_fact_data']['module'])
timestamps = Fact.get_host_timeline(hostname=self.builder.get_hostname(0), module='packages')
self.assertIsNotNone(timestamps)
self.assertEqual(len(timestamps), len(self.timestamps))
for i in range(0, self.scans):
self.assertEqual(timestamps[i], self.timestamps[i])
self.assertEqual(len(timestamps), self.builder.get_scan_count())
for i in range(0, self.builder.get_scan_count()):
self.assertEqual(timestamps[i], self.builder.get_timestamp(i))

90
awx/fact/tests/models/fact/fact_transform.py
View File

@@ -13,38 +13,45 @@ import pymongo
# AWX
from awx.fact.models.fact import * # noqa
from .base import BaseFactTest
from awx.fact.tests.base import BaseFactTest
__all__ = ['FactTransformTest', 'FactTransformUpdateTest',]
TEST_FACT_DATA = {
'hostname': 'hostname1',
'add_fact_data': {
'timestamp': datetime.now(),
'host': None,
'module': 'packages',
'fact': {
"acpid3.4": [
{
"version": "1:2.0.21-1ubuntu2",
"deeper.key": "some_value"
}
],
"adduser.2": [
{
"source": "apt",
"version": "3.113+nmu3ubuntu3"
}
],
"what.ever." : {
"shallowish.key": "some_shallow_value"
}
},
TEST_FACT_PACKAGES_WITH_DOTS = [
{
"name": "acpid3.4",
"version": "1:2.0.21-1ubuntu2",
"deeper.key": "some_value"
},
{
"name": "adduser.2",
"source": "apt",
"version": "3.113+nmu3ubuntu3"
},
{
"what.ever." : {
"shallowish.key": "some_shallow_value"
}
}
}
# Strip off microseconds because mongo has less precision
BaseFactTest.normalize_timestamp(TEST_FACT_DATA)
]
TEST_FACT_PACKAGES_WITH_DOLLARS = [
{
"name": "acpid3$4",
"version": "1:2.0.21-1ubuntu2",
"deeper.key": "some_value"
},
{
"name": "adduser$2",
"source": "apt",
"version": "3.113+nmu3ubuntu3"
},
{
"what.ever." : {
"shallowish.key": "some_shallow_value"
}
}
]
class FactTransformTest(BaseFactTest):
def setUp(self):
super(FactTransformTest, self).setUp()
@@ -52,16 +59,16 @@ class FactTransformTest(BaseFactTest):
self.client = pymongo.MongoClient('localhost', 27017)
self.db2 = self.client[settings.MONGO_DB]
self.create_host_document(TEST_FACT_DATA)
self.timestamp = datetime.now().replace(microsecond=0)
def setup_create_fact_dot(self):
self.data = TEST_FACT_DATA
self.f = Fact(**TEST_FACT_DATA['add_fact_data'])
self.host = FactHost(hostname='hosty').save()
self.f = Fact(timestamp=self.timestamp, module='packages', fact=TEST_FACT_PACKAGES_WITH_DOTS, host=self.host)
self.f.save()
def setup_create_fact_dollar(self):
self.data = TEST_FACT_DATA
self.f = Fact(**TEST_FACT_DATA['add_fact_data'])
self.host = FactHost(hostname='hosty').save()
self.f = Fact(timestamp=self.timestamp, module='packages', fact=TEST_FACT_PACKAGES_WITH_DOLLARS, host=self.host)
self.f.save()
def test_fact_with_dot_serialized(self):
@@ -73,17 +80,18 @@ class FactTransformTest(BaseFactTest):
# Bypass mongoengine and pymongo transform to get record
f_dict = self.db2['fact'].find_one(q)
self.assertIn('acpid3\uff0E4', f_dict['fact'])
self.assertIn('what\uff0Eever\uff0E', f_dict['fact'][2])
def test_fact_with_dot_serialized_pymongo(self):
#self.setup_create_fact_dot()
host = FactHost(hostname='hosty').save()
f = self.db['fact'].insert({
'hostname': TEST_FACT_DATA['hostname'],
'fact': TEST_FACT_DATA['add_fact_data']['fact'],
'timestamp': TEST_FACT_DATA['add_fact_data']['timestamp'],
'host': TEST_FACT_DATA['add_fact_data']['host'].id,
'module': TEST_FACT_DATA['add_fact_data']['module']
'hostname': 'hosty',
'fact': TEST_FACT_PACKAGES_WITH_DOTS,
'timestamp': self.timestamp,
'host': host.id,
'module': 'packages',
})
q = {
@@ -91,7 +99,7 @@ class FactTransformTest(BaseFactTest):
}
# Bypass mongoengine and pymongo transform to get record
f_dict = self.db2['fact'].find_one(q)
self.assertIn('acpid3\uff0E4', f_dict['fact'])
self.assertIn('what\uff0Eever\uff0E', f_dict['fact'][2])
def test_fact_with_dot_deserialized_pymongo(self):
self.setup_create_fact_dot()
@@ -100,13 +108,13 @@ class FactTransformTest(BaseFactTest):
'_id': self.f.id
}
f_dict = self.db['fact'].find_one(q)
self.assertIn('acpid3.4', f_dict['fact'])
self.assertIn('what.ever.', f_dict['fact'][2])
def test_fact_with_dot_deserialized(self):
self.setup_create_fact_dot()
f = Fact.objects.get(id=self.f.id)
self.assertIn('acpid3.4', f.fact)
self.assertIn('what.ever.', f.fact[2])
class FactTransformUpdateTest(BaseFactTest):
pass

2
awx/fact/tests/models/fact/fact_transform_pymongo.py
View File

@@ -13,7 +13,7 @@ import pymongo
# AWX
from awx.fact.models.fact import * # noqa
from .base import BaseFactTest
from awx.fact.tests.base import BaseFactTest
__all__ = ['FactSerializePymongoTest', 'FactDeserializePymongoTest',]