Skip to content

Commit

Permalink
use Unicode as the schema type for Snap configs
Browse files Browse the repository at this point in the history
  • Loading branch information
@st3v3nmw
st3v3nmw committed Jan 22, 2024
1 parent 0a22efc commit 2d16db1
Show file tree
Hide file tree
Showing 5 changed files with 9 additions and 176 deletions.
3 changes: 2 additions & 1 deletion landscape/client/monitor/snapmonitor.py
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import json
import logging

from landscape.client import snap_http
Expand Down Expand Up @@ -33,7 +34,7 @@ def get_data(self):
except SnapdHttpException:
config = {}

snaps[i]["config"] = config
snaps[i]["config"] = json.dumps(config)

# We get a lot of extra info from snapd. To avoid caching it all
# or invalidating the cache on timestamp changes, we use Message
Expand Down
50 changes: 6 additions & 44 deletions landscape/client/monitor/tests/test_snapmonitor.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,46 +50,8 @@ def test_get_data_snapd_http_exception(self):
)

@patch("landscape.client.monitor.snapmonitor.snap_http")
def test_get_simple_snap_config(self, snap_http_mock):
"""Tests that we can get and coerce simple snap config."""
plugin = SnapMonitor()
self.monitor.add(plugin)

snap_http_mock.list.return_value = SnapdResponse(
"sync",
200,
"OK",
[
{
"name": "test-snap",
"revision": "1",
"confinement": "strict",
"version": "v1.0",
"id": "123",
}
],
)
snap_http_mock.get_conf.return_value = {"bar": "default", "baz": False}
plugin.exchange()

messages = self.mstore.get_pending_messages()

self.assertTrue(len(messages) > 0)
self.assertDictEqual(
messages[0]["snaps"]["installed"][0],
{
"name": "test-snap",
"revision": "1",
"confinement": "strict",
"version": "v1.0",
"id": "123",
"config": {"bar": "default", "baz": False},
},
)

@patch("landscape.client.monitor.snapmonitor.snap_http")
def test_get_complex_snap_config(self, snap_http_mock):
"""Tests that we can get and coerce complex snap config."""
def test_get_snap_config(self, snap_http_mock):
"""Tests that we can get and coerce snap config."""
plugin = SnapMonitor()
self.monitor.add(plugin)

Expand Down Expand Up @@ -124,9 +86,9 @@ def test_get_complex_snap_config(self, snap_http_mock):
"confinement": "strict",
"version": "v1.0",
"id": "123",
"config": {
"foo": {"baz": "default", "qux": [1, True, 2.0]},
"bar": "enabled",
},
"config": (
'{"foo": {"baz": "default", "qux": [1, true, 2.0]}, '
'"bar": "enabled"}'
),
},
)
55 changes: 0 additions & 55 deletions landscape/lib/schema.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,4 @@
"""A schema system. Yes. Another one!"""
from functools import singledispatchmethod

from twisted.python.compat import iteritems
from twisted.python.compat import long
from twisted.python.compat import unicode
Expand Down Expand Up @@ -235,56 +233,3 @@ def coerce(self, value):
for k, v in value.items():
new_dict[self.key_schema.coerce(k)] = self.value_schema.coerce(v)
return new_dict


class Nested:
"""A type that allows nesting of C{list} and C{dict}.
@param schema: A `List`, `Dict`, or `KeyDict` schema.
"""

def __init__(self, schema):
if not isinstance(schema, (List, Dict, KeyDict)):
raise InvalidError(f"{schema} does not support nesting")

self.schema = schema

@singledispatchmethod
def coerce(self, value):
raise InvalidError(
f"{value!r} could not coerce with {self.schema}: "
"type does not support nesting"
)

@coerce.register(list)
def _(self, value):
if not isinstance(self.schema, List):
raise InvalidError(
f"{value!r} has type list which doesn't match {self.schema}"
)

new_sequence = []
for subvalue in value:
if isinstance(subvalue, list):
coerced = Nested(self.schema).coerce(subvalue)
else:
coerced = self.schema.schema.coerce(subvalue)
new_sequence.append(coerced)
return new_sequence

@coerce.register(dict)
def _(self, value):
if not isinstance(self.schema, Dict):
raise InvalidError(
f"{value!r} has type dict which doesn't match {self.schema}"
)

new_dict = {}
for k, v in value.items():
coerced_key = self.schema.key_schema.coerce(k)
if isinstance(v, dict):
coerced_value = Nested(self.schema).coerce(v)
else:
coerced_value = self.schema.value_schema.coerce(v)
new_dict[coerced_key] = coerced_value
return new_dict
52 changes: 0 additions & 52 deletions landscape/lib/tests/test_schema.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
from landscape.lib.schema import InvalidError
from landscape.lib.schema import KeyDict
from landscape.lib.schema import List
from landscape.lib.schema import Nested
from landscape.lib.schema import Tuple
from landscape.lib.schema import Unicode

Expand Down Expand Up @@ -226,54 +225,3 @@ def test_dict_inner_bad(self):

def test_dict_wrong_type(self):
self.assertRaises(InvalidError, Dict(Int(), Int()).coerce, 32)

def test_nested_list_correct_types_only(self):
schema = Nested(List(Any(Int(), Unicode())))
data = [[1, "foo"], [[5]]]
self.assertEqual(schema.coerce(data), [[1, "foo"], [[5]]])

def test_nested_list_bad_types_only(self):
schema = Nested(List(Any(Int(), Unicode())))
data = [[5.0]]
self.assertRaises(InvalidError, schema.coerce, data)

def test_nested_list_mixed_correct_and_bad_types_only(self):
schema = Nested(List(Any(Int(), Unicode())))
data = [[1.0, "foo"], [[5]]]
self.assertRaises(InvalidError, schema.coerce, data)

def test_nested_list_with_dictionary(self):
schema = Nested(List(Any(Int(), Unicode())))
data = {"foo": "bar"}
self.assertRaises(InvalidError, schema.coerce, data)

def test_nested_dict_correct_types_only(self):
schema = Nested(Dict(Unicode(), Any(Int(), Unicode())))
data = {"foo": 5, "bar": {"baz": "default"}}
self.assertEqual(
schema.coerce(data), {"foo": 5, "bar": {"baz": "default"}}
)

def test_nested_dict_bad_types_only(self):
schema = Nested(Dict(Unicode(), Any(Int(), Unicode())))
data = {2: 5.0}
self.assertRaises(InvalidError, schema.coerce, data)

def test_nested_dict_mixed_correct_and_bad_types_only(self):
schema = Nested(Dict(Unicode(), Any(Int(), Unicode())))
data = {"bar": {"baz": "default"}, 2: 5.0}
self.assertRaises(InvalidError, schema.coerce, data)

def test_nested_dict_with_list(self):
schema = Nested(Dict(Unicode(), Any(Int(), Unicode())))
data = []
self.assertRaises(InvalidError, schema.coerce, data)

def test_nested_with_type_that_doesnt_support_nesting(self):
with self.assertRaises(InvalidError):
Nested(Int())

def test_nested_coerce_unsupported_type(self):
schema = Nested(List(Any(Int(), Unicode())))
data = 4
self.assertRaises(InvalidError, schema.coerce, data)
25 changes: 1 addition & 24 deletions landscape/message_schemas/server_bound.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
from landscape.lib.schema import Int
from landscape.lib.schema import KeyDict
from landscape.lib.schema import List
from landscape.lib.schema import Nested
from landscape.lib.schema import Tuple
from landscape.lib.schema import Unicode

Expand Down Expand Up @@ -759,28 +758,6 @@
{"ubuntu-pro-reboot-required": Unicode()},
)

snap_config_vals = Nested(
Dict(
Unicode(),
Any(
Unicode(),
Int(),
Bool(),
Float(),
Nested(
List(
Any(
Unicode(),
Int(),
Bool(),
Float(),
),
)
),
),
),
)

SNAPS = Message(
"snaps",
{
Expand All @@ -804,7 +781,7 @@
),
"confinement": Unicode(),
"summary": Unicode(),
"config": snap_config_vals,
"config": Unicode(),
},
strict=False,
optional=[
Expand Down

0 comments on commit 2d16db1

Please sign in to comment.