blob: abfce004cb2c2d249baf952959670143e9d748b9 [file] [log] [blame]
# Copyright 2017-present Open Networking Foundation
#
# 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.
""" fake_stub.py
Implements a simple fake grpc stub to use for unit testing.
"""
import functools
ContentTypeMap = {}
class FakeObj(object):
BASES = []
def __init__(self, fields=[], **kwargs):
super(FakeObj, self).__setattr__("is_set", {})
super(FakeObj, self).__setattr__("fields", [])
for f in fields:
name = f["name"]
self.fields.append(name)
setattr(self, name, f["default"])
super(FakeObj, self).__setattr__("is_set", {})
for (k, v) in kwargs.items():
setattr(self, k, v)
def __repr__(self):
lines = []
for k in self.fields:
if self.is_set.get(k, False):
lines.append('%s: "%s"' % (k, getattr(self, k)))
if lines:
return "\n".join(lines) + "\n"
else:
return ""
def __setattr__(self, name, value):
self.is_set[name] = True
super(FakeObj, self).__setattr__(name, value)
def HasField(self, name):
""" Return True if the field is set in the protobuf. """
# gRPC throws a valueerror if the field doesn't exist in the schema
if name not in self.fields:
raise ValueError("Field %s does not exist in schema" % name)
# Fields that are always set
if name in ["leaf_model_name"]:
return True
field = self.DESCRIPTOR.fields_by_name[name].field_decl
# Reverse foreign keys lists are never None, they are an empty list
if field.get("fk_reverse", None):
return True
return self.is_set.get(name, False)
def ListFields(self):
fbn = self.DESCRIPTOR.fields_by_name
fieldlist = []
for (k, v) in fbn.items():
if self.is_set.get(k, False):
fieldlist.append((v, getattr(self, k)))
return fieldlist
@property
def self_content_type_id(self):
return "xos.%s" % self.__class__.__name__.lower()
class FakeExtensionManager(object):
def __init__(self, obj, extensions):
self.obj = obj
self.extensions = extensions
def _FindExtensionByName(self, name):
return name
def __getitem__(self, name, default=None):
if name in self.extensions:
return self.extensions[name]
return default
class FakeFieldOption(object):
def __init__(self, modelName=None, reverseFieldName=None):
self.modelName = modelName
self.reverseFieldName = reverseFieldName
class FakeField(object):
def __init__(self, field):
extensions = {}
self.field_decl = field
self.name = field["name"]
fk_model = field.get("fk_model", None)
if fk_model:
reverseFieldName = field.get("fk_reverseFieldName", None)
extensions["xos.foreignKey"] = FakeFieldOption(
modelName=fk_model, reverseFieldName=reverseFieldName
)
fk_reverse = field.get("fk_reverse", None)
if fk_reverse:
extensions["xos.reverseForeignKey"] = FakeFieldOption(modelName=fk_reverse)
self.Extensions = FakeExtensionManager(self, extensions)
def GetOptions(self):
return self
class FakeDescriptor(object):
def __init__(self, objName):
global ContentTypeIdCounter
global ContentTypeMap
self.objName = objName
if objName in ContentTypeMap:
ct = ContentTypeMap[objName]
else:
ct = "xos.%s" % objName.lower()
ContentTypeMap[objName] = ct
self.Extensions = FakeExtensionManager(self, {"xos.contentTypeId": ct})
def GetOptions(self):
return self
@property
def fields_by_name(self):
cls = globals()[self.objName]
fbn = {}
for field in cls.FIELDS:
fake_field = FakeField(field)
fbn[field["name"]] = fake_field
return fbn
class Controller(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{"name": "deployment_id", "default": 0, "fk_model": "Deployment"},
{"name": "class_names", "default": "Controller"},
)
def __init__(self, **kwargs):
return super(Controller, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("Controller")
class Deployment(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{"name": "class_names", "default": "Deployment"},
)
def __init__(self, **kwargs):
return super(Deployment, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("Controller")
class User(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "email", "default": ""},
{"name": "site_id", "default": 0, "fk_model": "Site"},
{"name": "class_names", "default": "User"},
)
def __init__(self, **kwargs):
return super(User, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("User")
class Slice(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{
"name": "site_id",
"default": 0,
"fk_model": "Site",
"fk_reverseFieldName": "slices",
},
{"name": "service_id", "default": 0, "fk_model": "Service"},
{"name": "creator_id", "default": 0, "fk_model": "User"},
{"name": "networks_ids", "default": [], "fk_reverse": "Network"},
{"name": "network", "default": ""},
{"name": "leaf_model_name", "default": "Slice"},
{"name": "class_names", "default": "Slice"},
)
def __init__(self, **kwargs):
return super(Slice, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("Slice")
class Site(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{"name": "login_base", "default": ""},
{"name": "slices_ids", "default": [], "fk_reverse": "Slice"},
{"name": "leaf_model_name", "default": "Site"},
{"name": "class_names", "default": "Site"},
)
def __init__(self, **kwargs):
return super(Site, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("Site")
class Service(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{"name": "slices_ids", "default": [], "fk_reverse": "Slice"},
{"name": "leaf_model_name", "default": "Service"},
{"name": "class_names", "default": "Service"},
)
def __init__(self, **kwargs):
return super(Service, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("Service")
class ServiceInstance(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "owher", "default": 0, "fk_model": "Service"},
{"name": "leaf_model_name", "default": "ServiceInstance"},
{"name": "class_names", "default": "ServiceInstance"},
)
def __init__(self, **kwargs):
return super(ServiceInstance, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("ServiceInstance")
class ONOSService(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{"name": "leaf_model_name", "default": "ONOSService"},
{"name": "class_names", "default": "ONOSService,Service"},
)
BASES = ["Service"]
def __init__(self, **kwargs):
return super(ONOSService, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("ONOSService")
class Network(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{"name": "owner_id", "default": 0, "fk_model": "Slice"},
{"name": "template_id", "default": 0, "fk_model": "NetworkTemplate"},
{
"name": "controllernetworks_ids",
"default": [],
"fk_reverse": "ControllerNetwork",
},
{"name": "leaf_model_name", "default": "Network"},
{"name": "class_names", "default": "Network"},
)
def __init__(self, **kwargs):
return super(Network, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("Network")
class NetworkTemplate(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "name", "default": ""},
{"name": "vtn_kind", "default": ""},
{"name": "leaf_model_name", "default": "NetworkTemplate"},
{"name": "class_names", "default": "NetworkTemplate"},
)
def __init__(self, **kwargs):
return super(NetworkTemplate, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("NetworkTemplate")
class ControllerNetwork(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "network_id", "default": 0, "fk_model": "Network"},
{"name": "controller_id", "default": 0, "fk_model": "Controller"},
{"name": "leaf_model_name", "default": "ControllerNetwork"},
{"name": "class_names", "default": "ControllerNetwork"},
)
def __init__(self, **kwargs):
return super(ControllerNetwork, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("ControllerNetwork")
class NetworkSlice(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "network_id", "default": 0, "fk_model": "Network"},
{"name": "slice_id", "default": 0, "fk_model": "Slice"},
{"name": "leaf_model_name", "default": "NetworkSlice"},
{"name": "class_names", "default": "NetworkSlice"},
)
def __init__(self, **kwargs):
return super(NetworkSlice, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("NetworkSlice")
class Tag(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "service_id", "default": None},
{"name": "name", "default": ""},
{"name": "value", "default": ""},
{"name": "content_type", "default": None},
{"name": "object_id", "default": None},
{"name": "leaf_model_name", "default": "Tag"},
{"name": "class_names", "default": "Tag"},
)
def __init__(self, **kwargs):
return super(Tag, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("Tag")
class TestModel(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "intfield", "default": 0},
{"name": "stringfield", "default": "somestring"},
{"name": "testmodeltwos_ids", "default": [], "fk_reverse": "TestModelTwo"},
{"name": "class_names", "default": "TestModel"},
)
def __init__(self, **kwargs):
return super(TestModel, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("TestModel")
class TestModelTwo(FakeObj):
FIELDS = (
{"name": "id", "default": 0},
{"name": "intfieldtwo", "default": 0},
{"name": "stringfieldtwo", "default": "somestringtwo"},
{
"name": "testmodel_id",
"default": 0,
"fk_model": "TestModel",
"fk_reverseFieldName": "testmodeltwos",
},
{"name": "class_names", "default": "TestModel"},
)
def __init__(self, **kwargs):
return super(TestModelTwo, self).__init__(self.FIELDS, **kwargs)
DESCRIPTOR = FakeDescriptor("TestModelTwo")
class ID(FakeObj):
pass
class FakeItemList(object):
def __init__(self, items):
self.items = items
class FakeElement(object):
EQUAL = "equal"
IEXACT = "iexact"
def __init__(self):
pass
class FakeElements(object):
def __init__(self):
self.items = []
def add(self):
el = FakeElement()
self.items.append(el)
return el
def __getitem__(self, index):
return self.items[index]
def __len__(self):
return len(self.items)
class FakeQuery(object):
DEFAULT = 0
ALL = 1
SYNCHRONIZER_DIRTY_OBJECTS = 2
SYNCHRONIZER_DELETED_OBJECTS = 3
SYNCHRONIZER_DIRTY_POLICIES = 4
SYNCHRONIZER_DELETED_POLICIES = 5
def __init__(self):
self.elements = FakeElements()
class FakeStub(object):
def __init__(self):
self.id_counter = 1
self.objs = {}
self.deleted_objs = {}
for name in [
"Controller",
"Deployment",
"Slice",
"Site",
"Tag",
"Service",
"ServiceInstance",
"ONOSService",
"User",
"Network",
"NetworkTemplate",
"ControllerNetwork",
"NetworkSlice",
"TestModel",
"TestModelTwo",
]:
setattr(self, "Get%s" % name, functools.partial(self.get, name))
setattr(self, "List%s" % name, functools.partial(self.list, name))
setattr(self, "Create%s" % name, functools.partial(self.create, name))
setattr(self, "Delete%s" % name, functools.partial(self.delete, name))
setattr(self, "Update%s" % name, functools.partial(self.update, name))
setattr(self, "Filter%s" % name, functools.partial(self.filter, name))
def make_key(self, name, id):
return "%s:%d" % (name, id.id)
def get(self, classname, id, metadata=None):
obj = self.objs.get(self.make_key(classname, id), None)
return obj
def list(self, classname, empty, metadata=None):
items = []
for (k, v) in self.objs.items():
(this_classname, id) = k.split(":")
if this_classname == classname:
items.append(v)
return FakeItemList(items)
def filter(self, classname, query, metadata=None):
items = []
if query.kind == FakeQuery.SYNCHRONIZER_DELETED_OBJECTS:
objs = self.deleted_objs.items()
else:
objs = self.objs.items()
for (k, v) in objs:
(this_classname, id) = k.split(":")
if this_classname != classname:
continue
match = True
for q in query.elements.items:
iValue = getattr(q, "iValue", None)
if (iValue is not None) and getattr(v, q.name) != iValue:
match = False
sValue = getattr(q, "sValue", None)
if (sValue is not None) and getattr(v, q.name) != sValue:
match = False
if match:
items.append(v)
return FakeItemList(items)
def create(self, classname, obj, metadata=None):
obj.id = self.id_counter
self.id_counter = self.id_counter + 1
k = self.make_key(classname, FakeObj(id=obj.id))
self.objs[k] = obj
for base_classname in obj.BASES:
base_class = globals()[base_classname]
base_obj = base_class(id=obj.id, leaf_model_name=classname)
k = self.make_key(base_classname, base_obj)
self.objs[k] = base_obj
return obj
def update(self, classname, obj, metadata=None):
# TODO: partial update support?
k = self.make_key(classname, FakeObj(id=obj.id))
self.objs[k] = obj
return obj
def delete(self, classname, id, metadata=None):
k = self.make_key(classname, id)
obj = self.objs[k]
del self.objs[k]
self.deleted_objs[k] = obj
class FakeCommonProtos(object):
def __init__(self):
self.ID = ID
self.Query = FakeQuery
class FakeProtos(object):
def __init__(self):
for name in [
"Controller",
"Deployment",
"Slice",
"Site",
"ID",
"Tag",
"Service",
"ServiceInstance",
"ONOSService",
"User",
"Network",
"NetworkTemplate",
"ControllerNetwork",
"NetworkSlice",
"TestModel",
"TestModelTwo",
]:
setattr(self, name, globals()[name])
self.common__pb2 = FakeCommonProtos()
class FakeSymDb(object):
def __init__(self):
self._classes = {}
for name in [
"Controller",
"Deployment",
"Slice",
"Site",
"ID",
"Tag",
"Service",
"ServiceInstance",
"ONOSService",
"User",
"Network",
"NetworkTemplate",
"ControllerNetwork",
"NetworkSlice",
"TestModel",
"TestModelTwo",
]:
self._classes["xos.%s" % name] = globals()[name]