def test__two_models_simple_ref(self):
self.use_models({
'model_one': 'select * from events',
'model_two': "select * from {{ref('model_one')}}",
})
compiler = self.get_compiler(self.get_project())
graph, linker = compiler.compile()
six.assertCountEqual(self,
linker.nodes(),
[
'model.test_models_compile.model_one',
'model.test_models_compile.model_two',
])
six.assertCountEqual(
self,
linker.edges(),
[ ('model.test_models_compile.model_one','model.test_models_compile.model_two',) ])
python类assertCountEqual()的实例源码
def test_subdir_uses_same_walker(self):
class CustomWalker(walk.Walker):
@classmethod
def bind(cls, fs):
return walk.BoundWalker(fs, walker_class=CustomWalker)
class CustomizedMemoryFS(MemoryFS):
walker_class=CustomWalker
base_fs=CustomizedMemoryFS()
base_fs.settext("a", "a")
base_fs.makedirs("b")
base_fs.settext("b/c", "c")
base_fs.settext("b/d", "d")
base_walker=base_fs.walk
self.assertEqual(base_walker.walker_class, CustomWalker)
six.assertCountEqual(self, ["/a", "/b/c", "/b/d"], base_walker.files())
sub_fs=base_fs.opendir("b")
sub_walker=sub_fs.walk
self.assertEqual(sub_walker.walker_class, CustomWalker)
six.assertCountEqual(self, ["/c", "/d"], sub_walker.files())
def test_dependency(self):
two_towers = """{
"name" : "test",
"layers" : {
"data": {
"parents": []
},
"conv1": {
"parents": ["data"]
},
"conv2": {
"parents": ["data"]
},
"output": {
"parents" : ["conv1", "conv2"]
}
}
}
"""
self.graph.load_from_string(two_towers)
nested_list = self._to_strings(self.graph.nested_list)
self.assertEqual(nested_list[0], 'data')
six.assertCountEqual(self, nested_list[1], (['conv1'], ['conv2']))
self.assertEqual(nested_list[2], 'output')
def test_dependency2(self):
two_towers = """{
"name" : "test",
"layers" : {
"data": {
"parents": []
},
"conv1": {
"parents": ["data"]
},
"output": {
"parents" : ["data", "conv1"]
}
}
}
"""
self.graph.load_from_string(two_towers)
nested_list = self._to_strings(self.graph.nested_list)
self.assertEqual(nested_list[0], 'data')
six.assertCountEqual(self, nested_list[1], (['conv1'], ))
self.assertEqual(nested_list[2], 'output')
def test_get_pdbs_for_gene(self):
model = 'e_coli_core'
gene = 'b0118'
expected = [('1l5j', 'A'), ('1l5j', 'B')]
six.assertCountEqual(self, expected, ssbio.databases.bigg.get_pdbs_for_gene(model, gene))
model = 'e_coli_core'
gene = 'b0351'
expected = []
six.assertCountEqual(self, expected, ssbio.databases.bigg.get_pdbs_for_gene(model, gene))
def test_finalize_stats_summaries(self):
p = plan.Plan(None)
p.save_summaries_secs = 42
p.losses['foo'] = tf.constant([1.0])
p.losses['bar'] = tf.constant([2.0, 3.0])
p.metrics['baz'] = tf.constant(4)
p.metrics['qux'] = tf.constant([5.0, 6.0])
p.finalize_stats()
with self.test_session():
self.assertEqual(6, p.loss_total.eval({p.batch_size_placeholder: 1}))
summary = tf.Summary()
summary.ParseFromString(p.summaries.eval({p.batch_size_placeholder: 1}))
qux_string = tf.summary.histogram('qux', [5, 6]).eval()
qux_proto = tf.Summary()
qux_proto.ParseFromString(qux_string)
qux_histogram = qux_proto.value[0].histo
expected_values = [
tf.Summary.Value(tag='foo', simple_value=1),
tf.Summary.Value(tag='bar', simple_value=5),
tf.Summary.Value(tag='loss_total', simple_value=6),
tf.Summary.Value(tag='baz', simple_value=4),
tf.Summary.Value(tag='qux', histo=qux_histogram)]
six.assertCountEqual(self, expected_values, summary.value)
summary.ParseFromString(p.summaries.eval({p.batch_size_placeholder: 2}))
expected_values = [
tf.Summary.Value(tag='foo', simple_value=0.5),
tf.Summary.Value(tag='bar', simple_value=2.5),
tf.Summary.Value(tag='loss_total', simple_value=3),
tf.Summary.Value(tag='baz', simple_value=4),
tf.Summary.Value(tag='qux', histo=qux_histogram)]
six.assertCountEqual(self, expected_values, summary.value)
def test_filestats(self):
path = 'netwerk/protocol/http/nsHttpConnectionMgr.cpp'
info = FileStats(path).get_info()
self.assertIsNotNone(info)
self.assertEqual(info['path'], 'netwerk/protocol/http/nsHttpConnectionMgr.cpp')
self.assertEqual(info['module'], 'Necko')
six.assertCountEqual(self, info['components'], ['Core::Networking', 'Core::Networking: Cache', 'Core::Networking: Cookies', 'Core::Networking: FTP', 'Core::Networking: File', 'Core::Networking: HTTP', 'Core::Networking: JAR', 'Core::Networking: Websockets'])
self.assertGreater(len(info['owners']), 0)
self.assertGreater(len(info['peers']), 0)
def test_compile_flags_boolean_simple(self):
flags = {
"--foo": True,
"--bar": False
}
compiled_args = pyqubes.compile.flags_boolean(flags)
six.assertCountEqual(self, compiled_args, ["--foo"])
def test_compile_flags_boolean_complex(self):
flags = {
"--foo": 1 != 1,
"--bar": "spam",
"--eggs": None,
}
compiled_args = pyqubes.compile.flags_boolean(flags)
six.assertCountEqual(self, compiled_args, ["--bar"])
def test_compile_flags_boolean_empty(self):
flags = {}
compiled_args = pyqubes.compile.flags_boolean(flags)
six.assertCountEqual(self, compiled_args, [])
def test_compile_flags_store_empty(self):
flags = {}
compiled_args = pyqubes.compile.flags_store(flags)
six.assertCountEqual(self, compiled_args, [])
def test_compile_flags_store_iterable_single_empty(self):
flags = {
"--foo": [],
}
compiled_args = pyqubes.compile.flags_store_iterable(flags)
six.assertCountEqual(self, compiled_args, [])
def test_constants_fedora_all(self):
six.assertCountEqual(self, c.FEDORA_ALL, [c.FEDORA, c.FEDORA_23])
def test_constants_debian_all(self):
six.assertCountEqual(self, c.DEBIAN_ALL, [c.DEBIAN, c.DEBIAN_8])
def test_vm_template_vm_create_app(self):
returned_vm = self.template_vm.create_app('app.thing')
six.assertCountEqual(self, self.enact_patch.call_args[0][0], ['qvm-create', 'app.thing', '--template', 'one.thing', '--label', 'red'])
self.assertIsInstance(returned_vm, pyqubes.vm.AppVM)
self.assertEqual(self.template_vm.proactive, returned_vm.proactive)
self.assertEqual(self.template_vm.operating_system, returned_vm.operating_system)
def test_vm_template_vm_create_app_flags(self):
returned_vm = self.template_vm.create_app('app.thing', label='green', standalone=True)
six.assertCountEqual(self, self.enact_patch.call_args[0][0], ['qvm-create', 'app.thing', '--standalone', '--template', 'one.thing', '--label', 'green'])
self.assertIsInstance(returned_vm, pyqubes.vm.AppVM)
self.assertEqual(self.template_vm.proactive, returned_vm.proactive)
self.assertEqual(self.template_vm.operating_system, returned_vm.operating_system)
def test_qvm_create_explicit(self):
command_args = pyqubes.qvm.qvm_create("pear", template='fruit', label='green')
six.assertCountEqual(self, command_args, ["qvm-create", "pear", "--template", "fruit", "--label", "green"])
def test_del_txt_record(self):
first_record_mock = mock.MagicMock()
first_record_mock.type = 'TXT'
first_record_mock.name = "DIFFERENT"
first_record_mock.data = self.record_content
correct_record_mock = mock.MagicMock()
correct_record_mock.type = 'TXT'
correct_record_mock.name = self.record_prefix
correct_record_mock.data = self.record_content
last_record_mock = mock.MagicMock()
last_record_mock.type = 'TXT'
last_record_mock.name = self.record_prefix
last_record_mock.data = "DIFFERENT"
domain_mock = mock.MagicMock()
domain_mock.name = DOMAIN
domain_mock.get_records.return_value = [first_record_mock,
correct_record_mock,
last_record_mock]
self.manager.get_all_domains.return_value = [domain_mock]
self.digitalocean_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
correct_record_mock.destroy.assert_called()
six.assertCountEqual(self, first_record_mock.destroy.call_args_list, [])
six.assertCountEqual(self, last_record_mock.destroy.call_args_list, [])
test_code_structure_checker.py 文件源码
项目:murano-pkg-check
作者: openstack
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_while_unknown_does(self):
MULTILINE_BODY = [
{'While': '$.deploy()',
'Does': ['$.a()', '$.b()']}
]
self.g = self._checker.codeblock(MULTILINE_BODY)
p1 = next(self.g)
p2 = next(self.g)
six.assertCountEqual(self, [
'Unknown keyword "Does" in "While"',
'Missing keyword "Do" for "While" code structure'],
[p1.message, p2.message])
def test__model_enabled(self):
self.use_models({
'model_one': 'select * from events',
'model_two': "select * from {{ref('model_one')}}",
})
cfg = {
"models": {
"materialized": "table",
"test_models_compile": {
"model_one": {"enabled": True},
"model_two": {"enabled": False},
}
}
}
compiler = self.get_compiler(self.get_project(cfg))
graph, linker = compiler.compile()
six.assertCountEqual(
self, linker.nodes(),
['model.test_models_compile.model_one',
'model.test_models_compile.model_two'])
six.assertCountEqual(
self, linker.edges(),
[('model.test_models_compile.model_one',
'model.test_models_compile.model_two',)])
def test_opendir(self):
# Make a simple directory structure
self.fs.makedir('foo')
self.fs.setbytes('foo/bar', b'barbar')
self.fs.setbytes('foo/egg', b'eggegg')
# Open a sub directory
with self.fs.opendir('foo') as foo_fs:
repr(foo_fs)
text_type(foo_fs)
six.assertCountEqual(self, foo_fs.listdir('/'), ['bar', 'egg'])
self.assertTrue(foo_fs.isfile('bar'))
self.assertTrue(foo_fs.isfile('egg'))
self.assertEqual(foo_fs.getbytes('bar'), b'barbar')
self.assertEqual(foo_fs.getbytes('egg'), b'eggegg')
self.assertFalse(self.fs.isclosed())
# Attempt to open a non-existent directory
with self.assertRaises(errors.ResourceNotFound):
self.fs.opendir('egg')
# Check error when doing opendir on a non dir
with self.assertRaises(errors.DirectoryExpected):
self.fs.opendir('foo/egg')
# These should work, and will essentially return a 'clone' of sorts
self.fs.opendir('')
self.fs.opendir('/')
# Check ClosingSubFS closes 'parent'
with self.fs.opendir('foo', factory=ClosingSubFS) as foo_fs:
six.assertCountEqual(self, foo_fs.listdir('/'), ['bar', 'egg'])
self.assertTrue(foo_fs.isfile('bar'))
self.assertTrue(foo_fs.isfile('egg'))
self.assertEqual(foo_fs.getbytes('bar'), b'barbar')
self.assertEqual(foo_fs.getbytes('egg'), b'eggegg')
self.assertTrue(self.fs.isclosed())
def test_get_attribute_list(self):
"""
Test that the attribute names of a managed object can be retrieved
with proper input.
"""
uid = 'b4faee10-aa2a-4446-8ad4-0881f3422959'
attribute_names = [
'Cryptographic Length',
'Cryptographic Algorithm',
'State',
'Digest',
'Lease Time',
'Initial Date',
'Unique Identifier',
'Name',
'Cryptographic Usage Mask',
'Object Type',
'Contact Information',
'Last Change Date']
result = results.GetAttributeListResult(
contents.ResultStatus(enums.ResultStatus.SUCCESS),
uid=uid,
names=attribute_names)
with ProxyKmipClient() as client:
client.proxy.get_attribute_list.return_value = result
result = client.get_attribute_list(uid)
client.proxy.get_attribute_list.assert_called_with(uid)
self.assertIsInstance(result, list)
six.assertCountEqual(self, attribute_names, result)
def test_headers_as_list(self):
six.assertCountEqual(self, self.uploader.headers_as_list,
["Tus-Resumable: 1.0.0"])
self.client.set_headers({'foo': 'bar'})
six.assertCountEqual(self, self.uploader.headers_as_list,
['Tus-Resumable: 1.0.0', 'foo: bar'])
def test_encode_metadata(self):
self.uploader.metadata = {'foo': 'bar', 'red': 'blue'}
encoded_metadata = ['foo' + ' ' + b64encode(b'bar').decode('ascii'),
'red' + ' ' + b64encode(b'blue').decode('ascii')]
six.assertCountEqual(self, self.uploader.encode_metadata(), encoded_metadata)
with pytest.raises(ValueError):
self.uploader.metadata = {'foo, ': 'bar'}
self.uploader.encode_metadata()
def test_get_attribute_list(self):
"""
Test that the attribute names of a managed object can be retrieved
with proper input.
"""
uid = 'b4faee10-aa2a-4446-8ad4-0881f3422959'
attribute_names = [
'Cryptographic Length',
'Cryptographic Algorithm',
'State',
'Digest',
'Lease Time',
'Initial Date',
'Unique Identifier',
'Name',
'Cryptographic Usage Mask',
'Object Type',
'Contact Information',
'Last Change Date']
result = results.GetAttributeListResult(
contents.ResultStatus(enums.ResultStatus.SUCCESS),
uid=uid,
names=attribute_names)
with ProxyKmipClient() as client:
client.proxy.get_attribute_list.return_value = result
result = client.get_attribute_list(uid)
client.proxy.get_attribute_list.assert_called_with(uid)
self.assertIsInstance(result, list)
six.assertCountEqual(self, attribute_names, result)
def test_no_device_token(self):
device_tokens = DeviceToken.objects.all()
six.assertCountEqual(self, device_tokens, [])
def test_no_resize_when_max_items_not_reached(self):
cd = LRUCache(max_items=2)
cd["a"] = "test-a"
self.assertEqual(len(cd), 1)
cd["b"] = "test-b"
self.assertEqual(len(cd), 2)
six.assertCountEqual(self, cd.keys(), ["a", "b"])
cd["c"] = "test-c"
self.assertEqual(len(cd), 2)
self.assertIn("c", cd)
def test_assertCountEqual():
class TestAssertCountEqual(unittest.TestCase):
def test(self):
with self.assertRaises(AssertionError):
six.assertCountEqual(self, (1, 2), [3, 4, 5])
six.assertCountEqual(self, (1, 2), [2, 1])
TestAssertCountEqual('test').test()
def test_replace_labels(self):
to_add, to_del = github.replace_labels(
self.component_titles, self.current_labels)
six.assertCountEqual(self, to_add, ['component:library'])
six.assertCountEqual(self, to_del, ['component:tool'])
def test_get_flavor_list(self):
uuids = [self.flavor['uuid']]
for i in range(1, 6):
flavor = utils.create_test_flavor(
uuid=uuidutils.generate_uuid(),
name=six.text_type(i))
uuids.append(six.text_type(flavor['uuid']))
res = self.dbapi.flavor_get_all(self.context)
res_uuids = [r['uuid'] for r in res]
six.assertCountEqual(self, uuids, res_uuids)