util: add matching_dicts
Move some of this logic to a common util, so it can be used in unittests. The curtin version is close but expects to work on a storage config, which is not a flat list.
This commit is contained in:
parent
2d74167614
commit
e0e856475c
|
@ -28,7 +28,10 @@ from unittest.mock import patch
|
||||||
from urllib.parse import unquote
|
from urllib.parse import unquote
|
||||||
|
|
||||||
from subiquitycore.tests import SubiTestCase
|
from subiquitycore.tests import SubiTestCase
|
||||||
from subiquitycore.utils import astart_command
|
from subiquitycore.utils import (
|
||||||
|
astart_command,
|
||||||
|
matching_dicts,
|
||||||
|
)
|
||||||
|
|
||||||
default_timeout = 10
|
default_timeout = 10
|
||||||
|
|
||||||
|
@ -37,8 +40,7 @@ def match(items, **kw):
|
||||||
typename = kw.pop('_type', None)
|
typename = kw.pop('_type', None)
|
||||||
if typename is not None:
|
if typename is not None:
|
||||||
kw['$type'] = typename
|
kw['$type'] = typename
|
||||||
return [item for item in items
|
return matching_dicts(items, **kw)
|
||||||
if all(item.get(k) == v for k, v in kw.items())]
|
|
||||||
|
|
||||||
|
|
||||||
def timeout(multiplier=1):
|
def timeout(multiplier=1):
|
||||||
|
|
|
@ -176,3 +176,10 @@ def disable_subiquity():
|
||||||
"snap.subiquity.subiquity-service.service",
|
"snap.subiquity.subiquity-service.service",
|
||||||
"serial-subiquity@*.service"])
|
"serial-subiquity@*.service"])
|
||||||
return
|
return
|
||||||
|
|
||||||
|
|
||||||
|
def matching_dicts(items, **kw):
|
||||||
|
"""Given an input sequence of dictionaries, return a list of dicts where
|
||||||
|
the supplied keyword arguments all match those items."""
|
||||||
|
return [item for item in items
|
||||||
|
if all(item.get(k) == v for k, v in kw.items())]
|
||||||
|
|
Loading…
Reference in New Issue