CORD-1311 support fake stub for unit test framework for ORM
Change-Id: Ifd5689d6ae67116f6ab258d4eb80204b2acba581
diff --git a/xos/xos_client/README.md b/xos/xos_client/README.md
index 1d45a2c..9757833 100644
--- a/xos/xos_client/README.md
+++ b/xos/xos_client/README.md
@@ -10,10 +10,14 @@
## Running Unit Tests ##
-Some unit tests (orm\_test.py) require an environment where the xos\_client library is installed, and a core API container is available to serve the API. It's suggested that the xos-client container be used together with a frontend or CiaB installation. For example,
+Some unit tests (orm\_test.py) optionally support an environment where the xos\_client library is installed, and a core API container is available to serve the API. This allows testing against the actual grpc client, instead of the mock-up. It's suggested that the xos-client container be used together with a frontend or CiaB installation.
docker run --rm -it --entrypoint bash docker-registry:5000/xosproject/xos-client:candidate
Once inside of the container, run the test(s). For example,
- python ./usr/local/lib/python2.7/dist-packages/xosapi/orm_test.py
+ python /usr/local/lib/python2.7/dist-packages/xosapi/orm_test.py -R
+
+The test may be run using a mock-up of the grpc client by omitting the -R option:
+
+ python orm-test.py
diff --git a/xos/xos_client/xosapi/fake_stub.py b/xos/xos_client/xosapi/fake_stub.py
new file mode 100644
index 0000000..6cfce8a
--- /dev/null
+++ b/xos/xos_client/xosapi/fake_stub.py
@@ -0,0 +1,125 @@
+""" fake_stub.py
+
+ Implements a simple fake grpc stub to use for unit testing.
+"""
+
+import functools
+
+ContentTypeIdCounter = 0;
+ContentTypeMap = {}
+
+class FakeObj(object):
+ def __init__(self, defaults={"id": 0}, **kwargs):
+ super(FakeObj, self).__setattr__("is_set", {})
+ super(FakeObj, self).__setattr__("fields", [])
+
+ for (k,v) in defaults.items():
+ self.fields.append(k)
+ setattr(self, k, v)
+
+ 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)
+
+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 FakeDescriptor(object):
+ def __init__(self, objName):
+ global ContentTypeIdCounter
+ global ContentTypeMap
+ if objName in ContentTypeMap:
+ ct = ContentTypeMap[objName]
+ else:
+ ct = ContentTypeIdCounter
+ ContentTypeIdCount = ContentTypeIdCounter + 1
+ ContentTypeMap[objName] = ct
+ self.Extensions = FakeExtensionManager(self, {"xos.contentTypeId": ct})
+
+ def GetOptions(self):
+ return self
+
+ @property
+ def fields_by_name(self):
+ # TODO: everything
+ return {}
+
+class Slice(FakeObj):
+ def __init__(self, **kwargs):
+ defaults = {"id": 0,
+ "name": ""}
+ return super(Slice, self).__init__(defaults, **kwargs)
+
+ DESCRIPTOR = FakeDescriptor("Slice")
+
+class FakeStub(object):
+ def __init__(self):
+ self.objs = {}
+ for name in ["Slice"]:
+ 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))
+
+
+ def make_key(self, name, id):
+ return "%s:%d" % (name, id.id)
+
+ def get(self, classname, id):
+ obj = self.objs.get(self.make_key(classname, id), None)
+ return obj
+
+ def list(self, classname, empty):
+ items = None
+ for (k,v) in self.objs.items():
+ (this_classname, id) = k.split(":")
+ if this_classname == classname:
+ items.append(v)
+ return items
+
+ def create(self, classname, obj):
+ k = self.make_key(classname, FakeObj(id=obj.id))
+ self.objs[k] = obj
+
+ def update(self, classname, obj):
+ # TODO: partial update support?
+ k = self.make_key(classname, FakeObj(id=obj.id))
+ self.objs[k] = obj
+
+ def delete(self, classname, id):
+ k = self.make_key(classname, id)
+ del self.objs[k]
+
+class FakeSymDb(object):
+ def __init__(self):
+ self._classes = {}
+ for name in ["Slice"]:
+ self._classes["xos.%s" % name] = globals()[name]
+
+
+
diff --git a/xos/xos_client/xosapi/orm.py b/xos/xos_client/xosapi/orm.py
index c4e1a3e..a030b6a 100644
--- a/xos/xos_client/xosapi/orm.py
+++ b/xos/xos_client/xosapi/orm.py
@@ -20,13 +20,8 @@
"""
import functools
-import grpc
-from google.protobuf.empty_pb2 import Empty
import time
-from google.protobuf import symbol_database as _symbol_database
-_sym_db = _symbol_database.Default()
-
convenience_wrappers = {}
class ORMWrapper(object):
@@ -305,10 +300,10 @@
return ORMQuerySet(result)
def all(self):
- return self.wrap_list(self._stub.invoke("List%s" % self._modelName, Empty()))
+ return self.wrap_list(self._stub.invoke("List%s" % self._modelName, self._stub.make_empty()))
def first(self):
- objs=self.wrap_list(self._stub.invoke("List%s" % self._modelName, Empty()))
+ objs=self.wrap_list(self._stub.invoke("List%s" % self._modelName, self._stub.make_empty()))
if not objs:
return None
return objs[0]
@@ -382,7 +377,7 @@
return self.objects.new(*args, **kwargs)
class ORMStub(object):
- def __init__(self, stub, package_name, invoker=None, caller_kind="grpcapi"):
+ def __init__(self, stub, package_name, invoker=None, caller_kind="grpcapi", sym_db = None, empty = None):
self.grpc_stub = stub
self.all_model_names = []
self.all_grpc_classes = {}
@@ -391,6 +386,17 @@
self.invoker = invoker
self.caller_kind = caller_kind
+ if not sym_db:
+ from google.protobuf import symbol_database as _symbol_database
+ sym_db = _symbol_database.Default()
+
+ self._sym_db = sym_db
+
+ if not empty:
+ from google.protobuf.empty_pb2 import Empty
+ empty = Empty
+ self._empty = empty
+
for name in dir(stub):
if name.startswith("Get"):
model_name = name[3:]
@@ -398,7 +404,7 @@
self.all_model_names.append(model_name)
- grpc_class = _sym_db._classes["%s.%s" % (package_name, model_name)]
+ grpc_class = self._sym_db._classes["%s.%s" % (package_name, model_name)]
self.all_grpc_classes[model_name] = grpc_class
ct = grpc_class.DESCRIPTOR.GetOptions().Extensions._FindExtensionByName("xos.contentTypeId")
@@ -451,10 +457,13 @@
raise
def make_ID(self, id):
- return _sym_db._classes["xos.ID"](id=id)
+ return self._sym_db._classes["xos.ID"](id=id)
+
+ def make_empty(self):
+ return self._empty()
def make_Query(self):
- return _sym_db._classes["xos.Query"]()
+ return self._sym_db._classes["xos.Query"]()
def listObjects(self):
return self.all_model_names
diff --git a/xos/xos_client/xosapi/orm_test.py b/xos/xos_client/xosapi/orm_test.py
index ed44435..39d13ad 100644
--- a/xos/xos_client/xosapi/orm_test.py
+++ b/xos/xos_client/xosapi/orm_test.py
@@ -3,52 +3,86 @@
import sys
import unittest
-from twisted.internet import reactor
-from xosapi import xos_grpc_client
+# Command-line argument of -R will cause this test to use a real grpc server
+# rather than the fake stub.
-exitStatus = -1
-
-# TODO: See if there's a way to stub this out using a fake xos_grpc_client
-# instead of the real one.
+if "-R" in sys.argv:
+ USE_FAKE_STUB = False
+ sys.argv.remove("-R")
+else:
+ USE_FAKE_STUB = True
class TestORM(unittest.TestCase):
+ def make_coreapi(self):
+ if USE_FAKE_STUB:
+ stub = FakeStub()
+ api = ORMStub(stub=stub, package_name = "xos", sym_db = FakeSymDb(), empty = FakeObj)
+ return api
+ else:
+ return xos_grpc_client.coreapi
+
def test_repr_name(self):
- s = xos_grpc_client.coreapi.Slice(name="foo")
+ orm = self.make_coreapi()
+ s = orm.Slice(name="foo")
self.assertNotEqual(s, None)
self.assertEqual(repr(s), "<Slice: foo>")
def test_str_name(self):
- s = xos_grpc_client.coreapi.Slice(name="foo")
+ orm = self.make_coreapi()
+ s = orm.Slice(name="foo")
self.assertNotEqual(s, None)
self.assertEqual(str(s), "foo")
def test_dumpstr_name(self):
- s = xos_grpc_client.coreapi.Slice(name="foo")
+ orm = self.make_coreapi()
+ s = orm.Slice(name="foo")
self.assertNotEqual(s, None)
self.assertEqual(s.dumpstr(), 'name: "foo"\n')
def test_repr_noname(self):
- s = xos_grpc_client.coreapi.Slice()
+ orm = self.make_coreapi()
+ s = orm.Slice()
self.assertNotEqual(s, None)
self.assertEqual(repr(s), "<Slice: id-0>")
def test_str_noname(self):
- s = xos_grpc_client.coreapi.Slice()
+ orm = self.make_coreapi()
+ s = orm.Slice()
self.assertNotEqual(s, None)
self.assertEqual(str(s), "Slice-0")
def test_dumpstr_noname(self):
- s = xos_grpc_client.coreapi.Slice()
+ orm = self.make_coreapi()
+ s = orm.Slice()
self.assertNotEqual(s, None)
self.assertEqual(s.dumpstr(), '')
-def test_callback():
- try:
- unittest.main()
- except exceptions.SystemExit, e:
- global exitStatus
- exitStatus = e.code
+if USE_FAKE_STUB:
+ sys.path.append("..")
-xos_grpc_client.start_api_parseargs(test_callback)
+ from fake_stub import FakeStub, FakeSymDb, FakeObj
+ from orm import ORMStub
-sys.exit(exitStatus)
+ print "Using Fake Stub"
+
+ unittest.main()
+else:
+ # This assumes xos-client python library is installed, and a gRPC server
+ # is available.
+
+ from twisted.internet import reactor
+ from xosapi import xos_grpc_client
+
+ print "Using xos-client library and core server"
+
+ def test_callback():
+ try:
+ unittest.main()
+ except exceptions.SystemExit, e:
+ global exitStatus
+ exitStatus = e.code
+
+ xos_grpc_client.start_api_parseargs(test_callback)
+
+ sys.exit(exitStatus)
+