diff options
Diffstat (limited to 'wsmeext/tests')
| -rw-r--r-- | wsmeext/tests/__init__.py | 0 | ||||
| -rw-r--r-- | wsmeext/tests/test_extdirect.py | 243 | ||||
| -rw-r--r-- | wsmeext/tests/test_soap.py | 423 | ||||
| -rw-r--r-- | wsmeext/tests/test_sqlalchemy_controllers.py | 223 | ||||
| -rw-r--r-- | wsmeext/tests/test_sqlalchemy_types.py | 72 |
5 files changed, 0 insertions, 961 deletions
diff --git a/wsmeext/tests/__init__.py b/wsmeext/tests/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/wsmeext/tests/__init__.py +++ /dev/null diff --git a/wsmeext/tests/test_extdirect.py b/wsmeext/tests/test_extdirect.py deleted file mode 100644 index 4c5bea8..0000000 --- a/wsmeext/tests/test_extdirect.py +++ /dev/null @@ -1,243 +0,0 @@ -import base64 -import datetime -import decimal - -try: - import simplejson as json -except ImportError: - import json # noqa - -import wsme.tests.protocol -from wsme.utils import parse_isodatetime, parse_isodate, parse_isotime -from wsme.types import isarray, isdict, isusertype - -import six - -if six.PY3: - from urllib.parse import urlencode -else: - from urllib import urlencode # noqa - - -def encode_arg(value): - if isinstance(value, tuple): - value, datatype = value - else: - datatype = type(value) - - if isinstance(datatype, list): - value = [encode_arg((item, datatype[0])) for item in value] - elif isinstance(datatype, dict): - key_type, value_type = list(datatype.items())[0] - value = dict(( - (encode_arg((key, key_type)), - encode_arg((value, value_type))) - for key, value in value.items() - )) - elif datatype in (datetime.date, datetime.time, datetime.datetime): - value = value.isoformat() - elif datatype == wsme.types.binary: - value = base64.encodestring(value).decode('ascii') - elif datatype == wsme.types.bytes: - value = value.decode('ascii') - elif datatype == decimal.Decimal: - value = str(value) - return value - - -def decode_result(value, datatype): - if value is None: - return None - if datatype == wsme.types.binary: - value = base64.decodestring(value.encode('ascii')) - return value - if isusertype(datatype): - datatype = datatype.basetype - if isinstance(datatype, list): - value = [decode_result(item, datatype[0]) for item in value] - elif isarray(datatype): - value = [decode_result(item, datatype.item_type) for item in value] - elif isinstance(datatype, dict): - key_type, value_type = list(datatype.items())[0] - value = dict(( - (decode_result(key, key_type), - decode_result(value, value_type)) - for key, value in value.items() - )) - elif isdict(datatype): - key_type, value_type = datatype.key_type, datatype.value_type - value = dict(( - (decode_result(key, key_type), - decode_result(value, value_type)) - for key, value in value.items() - )) - elif datatype == datetime.time: - value = parse_isotime(value) - elif datatype == datetime.date: - value = parse_isodate(value) - elif datatype == datetime.datetime: - value = parse_isodatetime(value) - elif hasattr(datatype, '_wsme_attributes'): - for attr in datatype._wsme_attributes: - if attr.key not in value: - continue - value[attr.key] = decode_result(value[attr.key], attr.datatype) - elif datatype == decimal.Decimal: - value = decimal.Decimal(value) - elif datatype == wsme.types.bytes: - value = value.encode('ascii') - elif datatype is not None and type(value) != datatype: - value = datatype(value) - return value - - -class TestExtDirectProtocol(wsme.tests.protocol.ProtocolTestCase): - protocol = 'extdirect' - protocol_options = { - 'namespace': 'MyNS.api', - 'nsfolder': 'app' - } - - def call(self, fname, _rt=None, _no_result_decode=False, _accept=None, - **kw): - path = fname.split('/') - try: - func, funcdef, args = self.root._lookup_function(path) - arguments = funcdef.arguments - except Exception: - arguments = [] - if len(path) == 1: - ns, action, fname = '', '', path[0] - elif len(path) == 2: - ns, action, fname = '', path[0], path[1] - else: - ns, action, fname = '.'.join(path[:-2]), path[-2], path[-1] - print(kw) - - args = [ - dict( - (arg.name, encode_arg(kw[arg.name])) - for arg in arguments if arg.name in kw - ) - ] - print("args =", args) - data = json.dumps({ - 'type': 'rpc', - 'tid': 0, - 'action': action, - 'method': fname, - 'data': args, - }) - print(data) - headers = {'Content-Type': 'application/json'} - if _accept: - headers['Accept'] = _accept - res = self.app.post('/extdirect/router/%s' % ns, data, headers=headers, - expect_errors=True) - - print(res.body) - - if _no_result_decode: - return res - - data = json.loads(res.text) - if data['type'] == 'rpc': - r = data['result'] - return decode_result(r, _rt) - elif data['type'] == 'exception': - faultcode, faultstring = data['message'].split(': ', 1) - debuginfo = data.get('where') - raise wsme.tests.protocol.CallException( - faultcode, faultstring, debuginfo) - - def test_api_alias(self): - assert self.root._get_protocol('extdirect').api_alias == '/app/api.js' - - def test_get_api(self): - res = self.app.get('/app/api.js') - print(res.body) - assert res.body - - def test_positional(self): - self.root._get_protocol('extdirect').default_params_notation = \ - 'positional' - - data = json.dumps({ - 'type': 'rpc', - 'tid': 0, - 'action': 'misc', - 'method': 'multiply', - 'data': [2, 5], - }) - headers = {'Content-Type': 'application/json'} - res = self.app.post('/extdirect/router', data, headers=headers) - - print(res.body) - - data = json.loads(res.text) - assert data['type'] == 'rpc' - r = data['result'] - assert r == 10 - - def test_batchcall(self): - data = json.dumps([{ - 'type': 'rpc', - 'tid': 1, - 'action': 'argtypes', - 'method': 'setdate', - 'data': [{'value': '2011-04-06'}], - }, { - 'type': 'rpc', - 'tid': 2, - 'action': 'returntypes', - 'method': 'getbytes', - 'data': [] - }]) - print(data) - headers = {'Content-Type': 'application/json'} - res = self.app.post('/extdirect/router', data, headers=headers) - - print(res.body) - - rdata = json.loads(res.text) - - assert len(rdata) == 2 - - assert rdata[0]['tid'] == 1 - assert rdata[0]['result'] == '2011-04-06' - assert rdata[1]['tid'] == 2 - assert rdata[1]['result'] == 'astring' - - def test_form_call(self): - params = { - 'value[0].inner.aint': 54, - 'value[1].inner.aint': 55, - 'extType': 'rpc', - 'extTID': 1, - 'extAction': 'argtypes', - 'extMethod': 'setnestedarray', - } - - body = urlencode(params) - r = self.app.post( - '/extdirect/router', - body, - headers={'Content-Type': 'application/x-www-form-urlencoded'} - ) - print(r) - - assert json.loads(r.text) == { - "tid": "1", - "action": "argtypes", - "type": "rpc", - "method": "setnestedarray", - "result": [{ - "inner": { - "aint": 54 - } - }, { - "inner": { - "aint": 55 - } - }] - } diff --git a/wsmeext/tests/test_soap.py b/wsmeext/tests/test_soap.py deleted file mode 100644 index bc17696..0000000 --- a/wsmeext/tests/test_soap.py +++ /dev/null @@ -1,423 +0,0 @@ -import decimal -import datetime -import base64 - -import six - -import wsme.tests.protocol - -try: - import xml.etree.ElementTree as et -except ImportError: - import cElementTree as et # noqa - -import suds.cache -import suds.client -import suds.transport - -import wsme.utils - - -class XDecimal(suds.xsd.sxbuiltin.XBuiltin): - def translate(self, value, topython=True): - if topython: - if isinstance(value, six.string_types) and len(value): - return decimal.Decimal(value) - else: - if isinstance(value, (decimal.Decimal, int, float)): - return str(value) - return value - - -suds.xsd.sxbuiltin.Factory.tags['decimal'] = XDecimal - - -class WebtestSudsTransport(suds.transport.Transport): - def __init__(self, app): - suds.transport.Transport.__init__(self) - self.app = app - - def open(self, request): - res = self.app.get(request.url, headers=request.headers) - return six.BytesIO(res.body) - - def send(self, request): - res = self.app.post( - request.url, - request.message, - headers=dict(( - (key, str(value)) for key, value in request.headers.items() - )), - expect_errors=True - ) - return suds.transport.Reply( - res.status_int, - dict(res.headers), - res.body - ) - - -class SudsCache(suds.cache.Cache): - def __init__(self): - self.d = {} - - def get(self, id): - return self.d.get(id) - - def getf(self, id): - b = self.get(id) - if b is not None: - return six.StringIO(self.get(id)) - - def put(self, id, bfr): - self.d[id] = bfr - - def putf(self, id, fp): - self.put(id, fp.read()) - - def purge(self, id): - try: - del self.d[id] - except KeyError: - pass - - def clear(self, id): - self.d = {} - - -sudscache = SudsCache() - -tns = "http://foo.bar.baz/soap/" -typenamespace = "http://foo.bar.baz/types/" - -soapenv_ns = 'http://schemas.xmlsoap.org/soap/envelope/' -xsi_ns = 'http://www.w3.org/2001/XMLSchema-instance' -body_qn = '{%s}Body' % soapenv_ns -fault_qn = '{%s}Fault' % soapenv_ns -faultcode_qn = '{%s}faultcode' % soapenv_ns -faultstring_qn = '{%s}faultstring' % soapenv_ns -faultdetail_qn = '{%s}detail' % soapenv_ns -type_qn = '{%s}type' % xsi_ns -nil_qn = '{%s}nil' % xsi_ns - - -def build_soap_message(method, params=""): - message = """<?xml version="1.0"?> -<soap:Envelope -xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/" -xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" -soap:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"> - - <soap:Body xmlns="%(typenamespace)s"> - <%(method)s> - %(params)s - </%(method)s> - </soap:Body> - -</soap:Envelope> -""" % dict(method=method, params=params, typenamespace=typenamespace) - return message - - -python_types = { - int: ('xs:int', str), - float: ('xs:float', str), - bool: ('xs:boolean', str), - wsme.types.bytes: ( - 'xs:string', - lambda x: x.decode('ascii') if isinstance(x, wsme.types.bytes) else x - ), - wsme.types.text: ('xs:string', wsme.types.text), - wsme.types.binary: ( - 'xs:base64Binary', - lambda x: base64.encodestring(x).decode('ascii') - ), - decimal.Decimal: ('xs:decimal', str), - datetime.date: ('xs:date', datetime.date.isoformat), - datetime.time: ('xs:time', datetime.time.isoformat), - datetime.datetime: ('xs:dateTime', datetime.datetime.isoformat), -} - -array_types = { - wsme.types.bytes: "String_Array", - wsme.types.text: "String_Array", - int: "Int_Array", - float: "Float_Array", - bool: "Boolean_Array", - datetime.datetime: "dateTime_Array" -} - -if not six.PY3: - array_types[long] = "Long_Array" # noqa - - -def tosoap(tag, value): - el = et.Element(tag) - if isinstance(value, tuple): - value, datatype = value - else: - datatype = type(value) - if value is None: - el.set('xsi:nil', 'true') - return el - if datatype in python_types: - stype, conv = python_types[datatype] - el.text = conv(value) - el.set('xsi:type', stype) - el.text = str(value) - return el - - -def tosuds(client, value): - if value is None: - return None - if isinstance(value, tuple): - value, datatype = value - else: - datatype = type(value) - if value is None: - return None - if isinstance(datatype, list): - if datatype[0] in array_types: - tname = array_types[datatype[0]] - else: - tname = datatype[0].__name__ + '_Array' - o = client.factory.create('types:' + tname) - o.item = [tosuds(client, (item, datatype[0])) for item in value] - return o - elif datatype in python_types: - return python_types[datatype][1](value) - else: - o = client.factory.create('types:' + datatype.__name__) - - for attr in datatype._wsme_attributes: - if attr.name in value: - setattr( - o, attr.name, - tosuds(client, (value[attr.name], attr.datatype)) - ) - return o - - -def read_bool(value): - return value == 'true' - - -soap_types = { - 'xs:string': wsme.types.text, - 'xs:int': int, - 'xs:long': int if six.PY3 else long, # noqa - 'xs:float': float, - 'xs:decimal': decimal.Decimal, - 'xs:boolean': read_bool, - 'xs:date': wsme.utils.parse_isodate, - 'xs:time': wsme.utils.parse_isotime, - 'xs:dateTime': wsme.utils.parse_isodatetime, - 'xs:base64Binary': base64.decodestring, -} - - -def fromsoap(el): - if el.get(nil_qn) == 'true': - return None - t = el.get(type_qn) - if t == 'xs:string': - return wsme.types.text(el.text if el.text else '') - if t in soap_types: - return soap_types[t](el.text) - elif t and t.endswith('_Array'): - return [fromsoap(i) for i in el] - else: - d = {} - for child in el: - name = child.tag - assert name.startswith('{%s}' % typenamespace), name - name = name[len(typenamespace) + 2:] - d[name] = fromsoap(child) - return d - - -def tobytes(value): - if isinstance(value, wsme.types.text): - value = value.encode() - return value - - -def tobin(value): - value = base64.decodestring(value.encode()) - return value - - -fromsuds_types = { - wsme.types.binary: tobin, - wsme.types.bytes: tobytes, - decimal.Decimal: decimal.Decimal, -} - - -def fromsuds(dt, value): - if value is None: - return None - if isinstance(dt, list): - return [fromsuds(dt[0], item) for item in value.item] - if wsme.types.isarray(dt): - return [fromsuds(dt.item_type, item) for item in value.item] - if wsme.types.isusertype(dt) and dt not in fromsuds_types: - dt = dt.basetype - if dt in fromsuds_types: - print(dt, value) - value = fromsuds_types[dt](value) - print(value) - return value - if wsme.types.iscomplex(dt): - d = {} - for attrdef in dt._wsme_attributes: - if not hasattr(value, attrdef.name): - continue - d[attrdef.name] = fromsuds( - attrdef.datatype, getattr(value, attrdef.name) - ) - return d - return value - - -class TestSOAP(wsme.tests.protocol.ProtocolTestCase): - protocol = 'soap' - protocol_options = dict(tns=tns, typenamespace=typenamespace) - ws_path = '/' - _sudsclient = None - - def setUp(self): - wsme.tests.protocol.ProtocolTestCase.setUp(self) - - def test_simple_call(self): - message = build_soap_message('touch') - print(message) - res = self.app.post( - self.ws_path, - message, - headers={"Content-Type": "application/soap+xml; charset=utf-8"}, - expect_errors=True - ) - print(res.body) - assert res.status.startswith('200') - - def call(self, fpath, _rt=None, _accept=None, _no_result_decode=False, - **kw): - - if _no_result_decode or _accept or self._testMethodName in ( - 'test_missing_argument', 'test_invalid_path', 'test_settext_empty', - 'test_settext_none' - ): - return self.raw_call(fpath, _rt, _accept, _no_result_decode, **kw) - - path = fpath.strip('/').split('/') - methodname = ''.join([path[0]] + [i.capitalize() for i in path[1:]]) - - m = getattr(self.sudsclient.service, methodname) - kw = dict(( - (key, tosuds(self.sudsclient, value)) for key, value in kw.items() - )) - print(kw) - try: - return fromsuds(_rt, m(**kw)) - except suds.WebFault as exc: - raise wsme.tests.protocol.CallException( - exc.fault.faultcode, - exc.fault.faultstring, - getattr(exc.fault, 'detail', None) or None - ) - - def raw_call(self, fpath, _rt=None, _accept=None, _no_result_decode=False, - **kw): - path = fpath.strip('/').split('/') - methodname = ''.join([path[0]] + [i.capitalize() for i in path[1:]]) - # get the actual definition so we can build the adequate request - if kw: - el = et.Element('parameters') - for key, value in kw.items(): - el.append(tosoap(key, value)) - - params = six.b("\n").join(et.tostring(el) for el in el) - else: - params = "" - methodname = ''.join([path[0]] + [i.capitalize() for i in path[1:]]) - message = build_soap_message(methodname, params) - print(message) - headers = {"Content-Type": "application/soap+xml; charset=utf-8"} - if _accept is not None: - headers['Accept'] = _accept - res = self.app.post( - self.ws_path, - message, - headers=headers, - expect_errors=True - ) - print("Status: ", res.status, "Received:", res.body) - - if _no_result_decode: - return res - - el = et.fromstring(res.body) - body = el.find(body_qn) - print(body) - - if res.status_int == 200: - response_tag = '{%s}%sResponse' % (typenamespace, methodname) - r = body.find(response_tag) - result = r.find('{%s}result' % typenamespace) - print("Result element: ", result) - return fromsoap(result) - elif res.status_int == 400: - fault = body.find(fault_qn) - raise wsme.tests.protocol.CallException( - fault.find(faultcode_qn).text, - fault.find(faultstring_qn).text, - "") - - elif res.status_int == 500: - fault = body.find(fault_qn) - raise wsme.tests.protocol.CallException( - fault.find(faultcode_qn).text, - fault.find(faultstring_qn).text, - fault.find(faultdetail_qn) is not None and - fault.find(faultdetail_qn).text or None) - - @property - def sudsclient(self): - if self._sudsclient is None: - self._sudsclient = suds.client.Client( - self.ws_path + 'api.wsdl', - transport=WebtestSudsTransport(self.app), - cache=sudscache - ) - return self._sudsclient - - def test_wsdl(self): - c = self.sudsclient - assert c.wsdl.tns[1] == tns, c.wsdl.tns - - sd = c.sd[0] - - assert len(sd.ports) == 1 - port, methods = sd.ports[0] - self.assertEqual(len(methods), 51) - - methods = dict(methods) - - assert 'returntypesGettext' in methods - print(methods) - - assert methods['argtypesSettime'][0][0] == 'value' - - def test_return_nesteddict(self): - pass - - def test_setnesteddict(self): - pass - - def test_return_objectdictattribute(self): - pass - - def test_setnested_nullobj(self): - pass # TODO write a soap adapted version of this test. diff --git a/wsmeext/tests/test_sqlalchemy_controllers.py b/wsmeext/tests/test_sqlalchemy_controllers.py deleted file mode 100644 index 1956788..0000000 --- a/wsmeext/tests/test_sqlalchemy_controllers.py +++ /dev/null @@ -1,223 +0,0 @@ -import datetime - -try: - import json -except ImportError: - import simplejson as json - -from webtest import TestApp - -from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy import Column, Integer, Unicode, Date, ForeignKey -from sqlalchemy.orm import relation - -from sqlalchemy import create_engine -from sqlalchemy.orm import sessionmaker, scoped_session - -from wsme import WSRoot -import wsme.types - -from wsmeext.sqlalchemy.types import generate_types -from wsmeext.sqlalchemy.controllers import CRUDController - -from six import u - -engine = create_engine('sqlite:///') -DBSession = scoped_session(sessionmaker(autocommit=False, autoflush=False, - bind=engine)) -DBBase = declarative_base() - -registry = wsme.types.Registry() - - -class DBPerson(DBBase): - __tablename__ = 'person' - - id = Column(Integer, primary_key=True) - name = Column(Unicode(50)) - birthdate = Column(Date) - - addresses = relation('DBAddress') - - -class DBAddress(DBBase): - __tablename__ = 'address' - - id = Column(Integer, primary_key=True) - - _person_id = Column('person_id', ForeignKey(DBPerson.id)) - - street = Column(Unicode(50)) - city = Column(Unicode(50)) - - person = relation(DBPerson) - - -globals().update( - generate_types(DBPerson, DBAddress, makename=lambda s: s[2:], - registry=registry)) - - -class PersonController(CRUDController): - __saclass__ = DBPerson - __dbsession__ = DBSession - __registry__ = registry - - -class AddressController(CRUDController): - __saclass__ = DBAddress - __dbsession__ = DBSession - __registry__ = registry - - -class Root(WSRoot): - __registry__ = registry - - person = PersonController() - address = AddressController() - - -class TestCRUDController(): - def setUp(self): - DBBase.metadata.create_all(DBSession.bind) - - self.root = Root() - self.root.getapi() - self.root.addprotocol('restjson') - - self.app = TestApp(self.root.wsgiapp()) - - def tearDown(self): - DBBase.metadata.drop_all(DBSession.bind) - - def test_create(self): - data = dict(data=dict( - name=u('Pierre-Joseph'), - birthdate=u('1809-01-15') - )) - r = self.app.post('/person/create', json.dumps(data), - headers={'Content-Type': 'application/json'}) - r = json.loads(r.text) - print(r) - assert r['name'] == u('Pierre-Joseph') - assert r['birthdate'] == u('1809-01-15') - - def test_PUT(self): - data = dict(data=dict( - name=u('Pierre-Joseph'), - birthdate=u('1809-01-15') - )) - r = self.app.put('/person', json.dumps(data), - headers={'Content-Type': 'application/json'}) - r = json.loads(r.text) - print(r) - assert r['name'] == u('Pierre-Joseph') - assert r['birthdate'] == u('1809-01-15') - - def test_read(self): - p = DBPerson( - name=u('Pierre-Joseph'), - birthdate=datetime.date(1809, 1, 15)) - DBSession.add(p) - DBSession.flush() - pid = p.id - r = self.app.post('/person/read', '{"ref": {"id": %s}}' % pid, - headers={'Content-Type': 'application/json'}) - r = json.loads(r.text) - print(r) - assert r['name'] == u('Pierre-Joseph') - assert r['birthdate'] == u('1809-01-15') - - def test_GET(self): - p = DBPerson( - name=u('Pierre-Joseph'), - birthdate=datetime.date(1809, 1, 15)) - DBSession.add(p) - DBSession.flush() - pid = p.id - r = self.app.get('/person?ref.id=%s' % pid, - headers={'Accept': 'application/json'}) - r = json.loads(r.text) - print(r) - assert r['name'] == u('Pierre-Joseph') - assert r['birthdate'] == u('1809-01-15') - - def test_GET_bad_accept(self): - p = DBPerson( - name=u('Pierre-Joseph'), - birthdate=datetime.date(1809, 1, 15)) - DBSession.add(p) - DBSession.flush() - pid = p.id - r = self.app.get('/person?ref.id=%s' % pid, - headers={'Accept': 'text/plain'}, - status=406) - assert r.text == ("Unacceptable Accept type: text/plain not in " - "['application/json', 'text/javascript', " - "'application/javascript', 'text/xml']") - - def test_update(self): - p = DBPerson( - name=u('Pierre-Joseph'), - birthdate=datetime.date(1809, 1, 15)) - DBSession.add(p) - DBSession.flush() - pid = p.id - data = { - "id": pid, - "name": u('Pierre-Joseph Proudon') - } - r = self.app.post('/person/update', json.dumps(dict(data=data)), - headers={'Content-Type': 'application/json'}) - r = json.loads(r.text) - print(r) - assert r['name'] == u('Pierre-Joseph Proudon') - assert r['birthdate'] == u('1809-01-15') - - def test_POST(self): - p = DBPerson( - name=u('Pierre-Joseph'), - birthdate=datetime.date(1809, 1, 15)) - DBSession.add(p) - DBSession.flush() - pid = p.id - data = { - "id": pid, - "name": u('Pierre-Joseph Proudon') - } - r = self.app.post('/person', json.dumps(dict(data=data)), - headers={'Content-Type': 'application/json'}) - r = json.loads(r.text) - print(r) - assert r['name'] == u('Pierre-Joseph Proudon') - assert r['birthdate'] == u('1809-01-15') - - def test_delete(self): - p = DBPerson( - name=u('Pierre-Joseph'), - birthdate=datetime.date(1809, 1, 15)) - DBSession.add(p) - DBSession.flush() - pid = p.id - r = self.app.post('/person/delete', json.dumps( - dict(ref=dict(id=pid))), - headers={ - 'Content-Type': 'application/json' - }) - print(r) - assert DBSession.query(DBPerson).get(pid) is None - - def test_DELETE(self): - p = DBPerson( - name=u('Pierre-Joseph'), - birthdate=datetime.date(1809, 1, 15)) - DBSession.add(p) - DBSession.flush() - pid = p.id - r = self.app.delete('/person?ref.id=%s' % pid, - headers={'Content-Type': 'application/json'}) - print(r) - assert DBSession.query(DBPerson).get(pid) is None - - def test_nothing(self): - pass diff --git a/wsmeext/tests/test_sqlalchemy_types.py b/wsmeext/tests/test_sqlalchemy_types.py deleted file mode 100644 index 8512015..0000000 --- a/wsmeext/tests/test_sqlalchemy_types.py +++ /dev/null @@ -1,72 +0,0 @@ -import datetime - -import wsmeext.sqlalchemy.types - -from wsme.types import text, Unset, isarray - -from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy import Column, Integer, String, Date, ForeignKey -from sqlalchemy.orm import relation - -from six import u - -SABase = declarative_base() - - -class SomeClass(SABase): - __tablename__ = 'some_table' - id = Column(Integer, primary_key=True) - name = Column(String(50)) - - adate = Column(Date) - - -def test_complextype(): - class AType(wsmeext.sqlalchemy.types.Base): - __saclass__ = SomeClass - - assert AType.id.datatype is int - assert AType.name.datatype is text - assert AType.adate.datatype is datetime.date - - a = AType() - s = SomeClass(name=u('aname'), adate=datetime.date(2012, 6, 26)) - assert s.name == u('aname') - - a.from_instance(s) - assert a.name == u('aname') - assert a.adate == datetime.date(2012, 6, 26) - - a.name = u('test') - del a.adate - assert a.adate is Unset - - a.to_instance(s) - assert s.name == u('test') - assert s.adate == datetime.date(2012, 6, 26) - - -def test_generate(): - class A(SABase): - __tablename__ = 'a' - id = Column(Integer, primary_key=True) - name = Column(String(50)) - - _b_id = Column(ForeignKey('b.id')) - - b = relation('B') - - class B(SABase): - __tablename__ = 'b' - id = Column(Integer, primary_key=True) - name = Column(String(50)) - - alist = relation(A) - - newtypes = wsmeext.sqlalchemy.types.generate_types(A, B) - - assert newtypes['A'].id.datatype is int - assert newtypes['A'].b.datatype is newtypes['B'] - assert newtypes['B'].id.datatype is int - assert isarray(newtypes['B'].alist.datatype) - assert newtypes['B'].alist.datatype.item_type is newtypes['A'] |