# See the License for the specific language governing permissions and
# limitations under the License.
# ==================================================================================
-from flask import Response
from jsonschema import validate
from jsonschema.exceptions import ValidationError
import connexion
"""
# mdc_logger.exception(exc) # waiting for https://jira.o-ran-sc.org/browse/RIC-39
return "", 503
- except BaseException:
- # catch all, should never happen...
- # mdc_logger.exception(exc) # waiting for https://jira.o-ran-sc.org/browse/RIC-39
- return Response(status=500)
+
+ # let other types of unexpected exceptions blow up and log
# Healthcheck
def find_and_get(self, prefix):
"""get all k v pairs that start with prefix"""
- ret_dict = self.sdl.find_and_get(A1NS, "{0}".format(prefix), atomic=True)
- found = {k: msgpack.unpackb(v, raw=False) for k, v in ret_dict.items()}
- # TODO: upgrade to sdl 2.0.0 which does the sorting for us
- return {k: found[k] for k in sorted(found)}
+ # note: SDL "*" usage is inconsistent with real python regex, where it would be ".*"
+ ret_dict = self.sdl.find_and_get(A1NS, "{0}*".format(prefix))
+ return {k: msgpack.unpackb(v, raw=False) for k, v in ret_dict.items()}
def delete(self, key):
""" delete a key"""
# The Jenkins job uses this string for the tag in the image name
# for example nexus3.o-ran-sc.org:10004/my-image-name:my-tag
---
-tag: 2.1.0
+tag: 2.1.1
:depth: 3
:local:
+[2.1.1] - 1/14/2020
+-------------------
+
+::
+
+ * Upgrades from sdl 1.0.0 to 2.0.2
+ * Delete a1test_helpers because SDL 2.0.2 provides the mockup we need
+ * Remove general catch all from A1
+
[2.1.0] - 1/8/2020
------------------
* Integration tests have a copy of a dbaas helm chart, however the goal is to simplify that deployment per https://jira.o-ran-sc.org/browse/RIC-45
* Unit tests have a mockup of SDL, however again the goal is to simplify as SDL grows per https://jira.o-ran-sc.org/browse/RIC-44
-
-
[2.0.0] - 12/9/2019
-------------------
apiVersion: v1
description: A1 Helm chart for Kubernetes
name: a1mediator
-version: 2.1.0
+version: 2.1.1
setup(
name="a1",
- version="2.1.0",
+ version="2.1.1",
packages=find_packages(exclude=["tests.*", "tests"]),
author="Tommy Carpenter",
description="RIC A1 Mediator for policy/intent changes",
"msgpack",
"rmr>=2.2.0",
"mdclogpy",
- "ricsdl>=1.0.2,<2.0.0",
+ "ricsdl>=2.0.2,<3.0.0",
],
package_data={"a1": ["openapi.yaml"]},
)
+++ /dev/null
-# ==================================================================================
-# Copyright (c) 2019-2020 Nokia
-# Copyright (c) 2018-2020 AT&T Intellectual Property.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-# ==================================================================================
-import msgpack
-from ricsdl.exceptions import RejectedByBackend, NotConnected, BackendError
-
-
-class MockSDLWrapper:
- """
- Mock wrapper for SDL that uses a dict so we do not rely on Redis for unit tests.
- Would be really nice if SDL itself came with a "standalone: dictionary" mode for this purpose...
- """
-
- def __init__(self):
- self.POLICY_DATA = {}
-
- def set(self, key, value):
- """set a key"""
-
- # these are for unit testing that the handler works on various SDL errors
- if key == "a1.policy_type.111":
- raise RejectedByBackend()
- if key == "a1.policy_type.112":
- raise NotConnected()
- if key == "a1.policy_type.113":
- raise BackendError()
-
- self.POLICY_DATA[key] = msgpack.packb(value, use_bin_type=True)
-
- def get(self, key):
- """get a key"""
- if key in self.POLICY_DATA:
- return msgpack.unpackb(self.POLICY_DATA[key], raw=False)
- return None
-
- def find_and_get(self, prefix):
- """get all k v pairs that start with prefix"""
- return {k: msgpack.unpackb(v, raw=False) for k, v in self.POLICY_DATA.items() if k.startswith(prefix)}
-
- def delete(self, key):
- """ delete a key"""
- del self.POLICY_DATA[key]
import time
import json
from rmr.rmr_mocks import rmr_mocks
+from ricsdl.syncstorage import SyncStorage
+from ricsdl.exceptions import RejectedByBackend, NotConnected, BackendError
from a1 import a1rmr, data
-from .a1test_helpers import MockSDLWrapper
RCV_ID = "test_receiver"
ADM_CRTL_TID = 6660666
def setup_module():
"""module level setup"""
- data.SDL = MockSDLWrapper() # patch SDL
+ data.SDL.sdl = SyncStorage(fake_db_backend="dict")
def noop():
pass
assert res.status_code == 204
# test 503 handlers
+
+ def monkey_set(key, value):
+ # set a key override function that throws sdl errors on certain keys
+ if key == "a1.policy_type.111":
+ raise RejectedByBackend()
+ if key == "a1.policy_type.112":
+ raise NotConnected()
+ if key == "a1.policy_type.113":
+ raise BackendError()
+
+ monkeypatch.setattr("a1.data.SDL.set", monkey_set)
+
res = client.put("/a1-p/policytypes/111", json=adm_type_good)
assert res.status_code == 503
res = client.put("/a1-p/policytypes/112", json=adm_type_good)
+"""
+tests data functions
+"""
# ==================================================================================
# Copyright (c) 2019-2020 Nokia
# Copyright (c) 2018-2020 AT&T Intellectual Property.
# limitations under the License.
# ==================================================================================
from a1 import data
-from .a1test_helpers import MockSDLWrapper
+from ricsdl.syncstorage import SyncStorage
def setup_module():
"""module level setup"""
- data.SDL = MockSDLWrapper() # patch SDL
+ data.SDL.sdl = SyncStorage(fake_db_backend="dict")
def test_sdl_raw():
Code Review / ric-plt / a1.git / commitdiff