1#!/usr/bin/env python 2# 3# spyne - Copyright (C) Spyne contributors. 4# 5# This library is free software; you can redistribute it and/or 6# modify it under the terms of the GNU Lesser General Public 7# License as published by the Free Software Foundation; either 8# version 2.1 of the License, or (at your option) any later version. 9# 10# This library is distributed in the hope that it will be useful, 11# but WITHOUT ANY WARRANTY; without even the implied warranty of 12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 13# Lesser General Public License for more details. 14# 15# You should have received a copy of the GNU Lesser General Public 16# License along with this library; if not, write to the Free Software 17# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 18# 19 20from __future__ import unicode_literals 21 22import logging 23logger = logging.getLogger(__name__) 24 25import unittest 26 27import uuid 28import pytz 29import decimal 30from spyne.util import six 31from spyne.util.dictdoc import get_object_as_dict 32 33if not six.PY2: 34 long = int 35 36from datetime import datetime 37from datetime import date 38from datetime import time 39from datetime import timedelta 40 41import lxml.etree 42import lxml.html 43 44from lxml.builder import E 45 46from spyne import MethodContext 47from spyne.service import Service 48from spyne.server import ServerBase 49from spyne.application import Application 50from spyne.decorator import srpc, rpc 51from spyne.error import ValidationError 52from spyne.model.binary import binary_encoding_handlers, File 53from spyne.model.complex import ComplexModel 54from spyne.model.complex import Iterable 55from spyne.model.fault import Fault 56from spyne.protocol import ProtocolBase 57from spyne.model.binary import ByteArray 58from spyne.model.primitive import Decimal 59from spyne.model.primitive import Integer 60from spyne.model.primitive import String 61from spyne.model.primitive import DateTime 62from spyne.model.primitive import Mandatory 63from spyne.model.primitive import AnyXml 64from spyne.model.primitive import AnyHtml 65from spyne.model.primitive import AnyDict 66from spyne.model.primitive import Unicode 67from spyne.model.primitive import AnyUri 68from spyne.model.primitive import ImageUri 69from spyne.model.primitive import Double 70from spyne.model.primitive import Integer8 71from spyne.model.primitive import Time 72from spyne.model.primitive import Date 73from spyne.model.primitive import Duration 74from spyne.model.primitive import Boolean 75from spyne.model.primitive import Uuid 76from spyne.model.primitive import Point 77from spyne.model.primitive import Line 78from spyne.model.primitive import Polygon 79from spyne.model.primitive import MultiPoint 80from spyne.model.primitive import MultiLine 81from spyne.model.primitive import MultiPolygon 82 83 84def _unbyte(d): 85 if d is None: 86 return 87 88 for k, v in list(d.items()): 89 if isinstance(k, bytes): 90 del d[k] 91 d[k.decode('utf8')] = v 92 93 if isinstance(v, dict): 94 _unbyte(v) 95 96 for k, v in d.items(): 97 if isinstance(v, (list, tuple)): 98 l = [] 99 for sub in v: 100 if isinstance(sub, dict): 101 l.append(_unbyte(sub)) 102 103 elif isinstance(sub, bytes): 104 l.append(sub.decode("utf8")) 105 106 else: 107 l.append(sub) 108 109 d[k] = tuple(l) 110 111 elif isinstance(v, bytes): 112 try: 113 d[k] = v.decode('utf8') 114 except UnicodeDecodeError: 115 d[k] = v 116 117 return d 118 119 120def TDry(serializer, _DictDocumentChild, dumps_kwargs=None): 121 if not dumps_kwargs: 122 dumps_kwargs = {} 123 124 def _dry_me(services, d, ignore_wrappers=False, complex_as=dict, 125 just_ctx=False, just_in_object=False, validator=None, 126 polymorphic=False): 127 128 app = Application(services, 'tns', 129 in_protocol=_DictDocumentChild( 130 ignore_wrappers=ignore_wrappers, complex_as=complex_as, 131 polymorphic=polymorphic, validator=validator, 132 ), 133 out_protocol=_DictDocumentChild( 134 ignore_wrappers=ignore_wrappers, complex_as=complex_as, 135 polymorphic=polymorphic), 136 ) 137 138 server = ServerBase(app) 139 initial_ctx = MethodContext(server, MethodContext.SERVER) 140 in_string = serializer.dumps(d, **dumps_kwargs) 141 if not isinstance(in_string, bytes): 142 in_string = in_string.encode('utf8') 143 initial_ctx.in_string = [in_string] 144 145 ctx, = server.generate_contexts(initial_ctx, in_string_charset='utf8') 146 if not just_ctx: 147 server.get_in_object(ctx) 148 if not just_in_object: 149 server.get_out_object(ctx) 150 server.get_out_string(ctx) 151 152 return ctx 153 return _dry_me 154 155def TDictDocumentTest(serializer, _DictDocumentChild, dumps_kwargs=None, 156 loads_kwargs=None, convert_dict=None): 157 if not dumps_kwargs: 158 dumps_kwargs = {} 159 if not loads_kwargs: 160 loads_kwargs = {} 161 _dry_me = TDry(serializer, _DictDocumentChild, dumps_kwargs) 162 163 if convert_dict is None: 164 convert_dict = lambda v: v 165 166 class Test(unittest.TestCase): 167 def dumps(self, o): 168 print("using", dumps_kwargs, "to dump", o) 169 return serializer.dumps(o, **dumps_kwargs) 170 171 def loads(self, o): 172 return _unbyte(serializer.loads(o, **loads_kwargs)) 173 174 def test_complex_with_only_primitive_fields(self): 175 class SomeComplexModel(ComplexModel): 176 i = Integer 177 s = Unicode 178 179 class SomeService(Service): 180 @srpc(SomeComplexModel, _returns=SomeComplexModel) 181 def some_call(scm): 182 return SomeComplexModel(i=5, s='5x') 183 184 ctx = _dry_me([SomeService], {"some_call":[]}) 185 186 s = self.loads(b''.join(ctx.out_string)) 187 188 s = s["some_callResponse"]["some_callResult"]["SomeComplexModel"] 189 assert s["i"] == 5 190 assert s["s"] in ("5x", b"5x") 191 192 def test_complex(self): 193 class CM(ComplexModel): 194 i = Integer 195 s = Unicode 196 197 class CCM(ComplexModel): 198 c = CM 199 i = Integer 200 s = Unicode 201 202 class SomeService(Service): 203 @srpc(CCM, _returns=CCM) 204 def some_call(ccm): 205 return CCM(c=ccm.c, i=ccm.i, s=ccm.s) 206 207 ctx = _dry_me([SomeService], {"some_call": 208 {"ccm": {"CCM":{ 209 "c":{"CM":{"i":3, "s": "3x"}}, 210 "i":4, 211 "s": "4x", 212 }}} 213 }) 214 215 ret = self.loads(b''.join(ctx.out_string)) 216 print(ret) 217 218 d = ret['some_callResponse']['some_callResult']['CCM'] 219 assert d['i'] == 4 220 assert d['s'] in ('4x', b'4x') 221 assert d['c']['CM']['i'] == 3 222 assert d['c']['CM']['s'] in ('3x', b'3x') 223 224 def test_multiple_list(self): 225 class SomeService(Service): 226 @srpc(Unicode(max_occurs=decimal.Decimal('inf')), 227 _returns=Unicode(max_occurs=decimal.Decimal('inf'))) 228 def some_call(s): 229 return s 230 231 ctx = _dry_me([SomeService], {"some_call":[["a","b"]]}) 232 233 data = b''.join(ctx.out_string) 234 print(data) 235 236 assert self.loads(data) == \ 237 {"some_callResponse": {"some_callResult": ("a", "b")}} 238 239 def test_multiple_dict(self): 240 class SomeService(Service): 241 @srpc(Unicode(max_occurs=decimal.Decimal('inf')), 242 _returns=Unicode(max_occurs=decimal.Decimal('inf'))) 243 def some_call(s): 244 return s 245 246 ctx = _dry_me([SomeService], {"some_call":{"s":["a","b"]}}) 247 248 assert self.loads(b''.join(ctx.out_string)) == \ 249 {"some_callResponse": {"some_callResult": ("a", "b")}} 250 251 def test_multiple_dict_array(self): 252 class SomeService(Service): 253 @srpc(Iterable(Unicode), _returns=Iterable(Unicode)) 254 def some_call(s): 255 return s 256 257 ctx = _dry_me([SomeService], {"some_call":{"s":["a","b"]}}) 258 259 assert self.loads(b''.join(ctx.out_string)) == \ 260 {"some_callResponse": {"some_callResult": ("a", "b")}} 261 262 def test_multiple_dict_complex_array(self): 263 class CM(ComplexModel): 264 i = Integer 265 s = Unicode 266 267 class CCM(ComplexModel): 268 c = CM 269 i = Integer 270 s = Unicode 271 272 class ECM(CCM): 273 d = DateTime 274 275 class SomeService(Service): 276 @srpc(Iterable(ECM), _returns=Iterable(ECM)) 277 def some_call(ecm): 278 return ecm 279 280 ctx = _dry_me([SomeService], { 281 "some_call": {"ecm": [{"ECM": { 282 "c": {"CM":{"i":3, "s": "3x"}}, 283 "i":4, 284 "s": "4x", 285 "d": "2011-12-13T14:15:16Z" 286 }}] 287 }}) 288 289 print(ctx.in_object) 290 291 ret = self.loads(b''.join(ctx.out_string)) 292 print(ret) 293 assert ret["some_callResponse"]['some_callResult'] 294 assert ret["some_callResponse"]['some_callResult'][0] 295 assert ret["some_callResponse"]['some_callResult'][0]["ECM"]["c"] 296 assert ret["some_callResponse"]['some_callResult'][0]["ECM"]["c"]["CM"]["i"] == 3 297 assert ret["some_callResponse"]['some_callResult'][0]["ECM"]["c"]["CM"]["s"] in ("3x", b"3x") 298 assert ret["some_callResponse"]['some_callResult'][0]["ECM"]["i"] == 4 299 assert ret["some_callResponse"]['some_callResult'][0]["ECM"]["s"] in ("4x", b"4x") 300 assert ret["some_callResponse"]['some_callResult'][0]["ECM"]["d"] == "2011-12-13T14:15:16+00:00" 301 302 def test_invalid_request(self): 303 class SomeService(Service): 304 @srpc(Integer, String, DateTime) 305 def yay(i,s,d): 306 print(i,s,d) 307 308 ctx = _dry_me([SomeService], {"some_call": {"yay": []}}, 309 just_in_object=True) 310 311 print(ctx.in_error) 312 assert ctx.in_error.faultcode == 'Client.ResourceNotFound' 313 314 def test_invalid_string(self): 315 class SomeService(Service): 316 @srpc(Integer, String, DateTime) 317 def yay(i,s,d): 318 print(i, s, d) 319 320 ctx = _dry_me([SomeService], {"yay": {"s": 1}}, validator='soft', 321 just_in_object=True) 322 323 assert ctx.in_error.faultcode == 'Client.ValidationError' 324 325 def test_invalid_number(self): 326 class SomeService(Service): 327 @srpc(Integer, String, DateTime) 328 def yay(i,s,d): 329 print(i,s,d) 330 331 ctx = _dry_me([SomeService], {"yay": ["s", "B"]}, validator='soft', 332 just_in_object=True) 333 334 assert ctx.in_error.faultcode == 'Client.ValidationError' 335 336 def test_missing_value(self): 337 class SomeService(Service): 338 @srpc(Integer, Unicode, Mandatory.DateTime) 339 def yay(i, s, d): 340 print(i, s, d) 341 342 ctx = _dry_me([SomeService], {"yay": [1, "B"]}, validator='soft', 343 just_in_object=True) 344 345 print(ctx.in_error.faultstring) 346 assert ctx.in_error.faultcode == 'Client.ValidationError' 347 assert ctx.in_error.faultstring.endswith("at least 1 times.") 348 349 def test_invalid_datetime(self): 350 class SomeService(Service): 351 @srpc(Integer, String, Mandatory.DateTime) 352 def yay(i,s,d): 353 print(i,s,d) 354 355 ctx = _dry_me([SomeService],{"yay": {"d":"a2011"}},validator='soft', 356 just_in_object=True) 357 358 assert ctx.in_error.faultcode == 'Client.ValidationError' 359 360 def test_fault_to_dict(self): 361 class SomeService(Service): 362 @srpc(_returns=String) 363 def some_call(): 364 raise Fault() 365 366 _dry_me([SomeService], {"some_call":[]}) 367 368 def test_prune_none_and_optional(self): 369 class SomeObject(ComplexModel): 370 i = Integer 371 s = String(min_occurs=1) 372 373 class SomeService(Service): 374 @srpc(_returns=SomeObject) 375 def some_call(): 376 return SomeObject() 377 378 ctx = _dry_me([SomeService], {"some_call":[]}) 379 380 ret = self.loads(b''.join(ctx.out_string)) 381 382 assert ret == {"some_callResponse": {'some_callResult': 383 {'SomeObject': {'s': None}}}} 384 385 def test_any_xml(self): 386 d = lxml.etree.tostring(E('{ns1}x', E('{ns2}Y', "some data")), 387 encoding='unicode') 388 389 class SomeService(Service): 390 @srpc(AnyXml, _returns=AnyXml) 391 def some_call(p): 392 print(p) 393 print(type(p)) 394 assert type(p) == lxml.etree._Element 395 return p 396 397 ctx = _dry_me([SomeService], {"some_call":[d]}) 398 399 s = self.loads(b''.join(ctx.out_string)) 400 d = {"some_callResponse": {"some_callResult": d}} 401 print(s) 402 print(d) 403 assert s == d 404 405 def test_any_html(self): 406 d = lxml.html.tostring(E('div', E('span', "something")), 407 encoding='unicode') 408 409 class SomeService(Service): 410 @srpc(AnyHtml, _returns=AnyHtml) 411 def some_call(p): 412 print(p) 413 print(type(p)) 414 assert type(p) == lxml.html.HtmlElement 415 return p 416 417 ctx = _dry_me([SomeService], {"some_call": [d]}) 418 419 s = self.loads(b''.join(ctx.out_string)) 420 d = {"some_callResponse": {"some_callResult": d}} 421 422 print(s) 423 print(d) 424 assert s == d 425 426 def test_any_dict(self): 427 d = {'helo': 213, 'data': {'nested': [12, 0.3]}} 428 429 class SomeService(Service): 430 @srpc(AnyDict, _returns=AnyDict) 431 def some_call(p): 432 print(p) 433 print(type(p)) 434 assert type(p) == dict 435 return p 436 437 ctx = _dry_me([SomeService], {"some_call":[d]}) 438 439 s = b''.join(ctx.out_string) 440 d = self.dumps({"some_callResponse": {"some_callResult": d}}) 441 442 print(s) 443 print(d) 444 assert self.loads(s) == self.loads(d) 445 446 def test_unicode(self): 447 d = u'some string' 448 449 class SomeService(Service): 450 @srpc(Unicode, _returns=Unicode) 451 def some_call(p): 452 print(p) 453 print(type(p)) 454 assert type(p) == six.text_type 455 return p 456 457 ctx = _dry_me([SomeService], {"some_call":[d]}) 458 459 s = self.loads(b''.join(ctx.out_string)) 460 d = {"some_callResponse": {"some_callResult": d}} 461 print(s) 462 print(d) 463 assert s == d 464 465 def test_any_uri(self): 466 d = 'http://example.com/?asd=b12&df=aa#tag' 467 468 class SomeService(Service): 469 @srpc(AnyUri, _returns=AnyUri) 470 def some_call(p): 471 print(p) 472 print(type(p)) 473 assert isinstance(p, six.string_types) 474 return p 475 476 ctx = _dry_me([SomeService], {"some_call": [d]}) 477 478 s = self.loads(b''.join(ctx.out_string)) 479 d = {"some_callResponse": {"some_callResult": d}} 480 print(s) 481 print(d) 482 assert s == d 483 484 def test_image_uri(self): 485 d = 'http://example.com/funny.gif' 486 487 class SomeService(Service): 488 @srpc(ImageUri, _returns=ImageUri) 489 def some_call(p): 490 print(p) 491 print(type(p)) 492 assert isinstance(p, six.string_types) 493 return p 494 495 ctx = _dry_me([SomeService], {"some_call": [d]}) 496 497 s = self.loads(b''.join(ctx.out_string)) 498 d = {"some_callResponse": {"some_callResult": d}} 499 print(s) 500 print(d) 501 assert s == d 502 503 def test_decimal(self): 504 d = decimal.Decimal('1e100') 505 if _DictDocumentChild._decimal_as_string: 506 d = str(d) 507 508 class SomeService(Service): 509 @srpc(Decimal, _returns=Decimal) 510 def some_call(p): 511 print(p) 512 print(type(p)) 513 assert type(p) == decimal.Decimal 514 return p 515 516 ctx = _dry_me([SomeService], {"some_call": [d]}) 517 518 s = self.loads(b''.join(ctx.out_string)) 519 d = {"some_callResponse": {"some_callResult": d}} 520 print(s) 521 print(d) 522 assert s == d 523 524 def test_double(self): 525 d = 12.3467 526 527 class SomeService(Service): 528 @srpc(Double, _returns=Double) 529 def some_call(p): 530 print(p) 531 print(type(p)) 532 assert type(p) == float 533 return p 534 535 ctx = _dry_me([SomeService], {"some_call":[d]}) 536 537 s = self.loads(b''.join(ctx.out_string)) 538 d = {"some_callResponse": {"some_callResult": d}} 539 print(s) 540 print(d) 541 assert s == d 542 543 def test_integer(self): 544 d = 5 545 546 class SomeService(Service): 547 @srpc(Integer, _returns=Integer) 548 def some_call(p): 549 print(p) 550 print(type(p)) 551 assert type(p) == int 552 return p 553 554 ctx = _dry_me([SomeService], {"some_call":[d]}) 555 556 s = self.loads(b''.join(ctx.out_string)) 557 d = {"some_callResponse": {"some_callResult": d}} 558 print(s) 559 print(d) 560 assert s == d 561 562 def test_integer_way_small(self): 563 d = -1<<1000 564 if _DictDocumentChild._huge_numbers_as_string: 565 d = str(d) 566 567 class SomeService(Service): 568 @srpc(Integer, _returns=Integer) 569 def some_call(p): 570 print(p) 571 print(type(p)) 572 assert type(p) == long 573 return p 574 575 ctx = _dry_me([SomeService], {"some_call":[d]}) 576 577 s = self.loads(b''.join(ctx.out_string)) 578 d = {"some_callResponse": {"some_callResult": d}} 579 580 print(s) 581 print(d) 582 assert s == d 583 584 def test_integer_way_big(self): 585 d = 1<<1000 586 if _DictDocumentChild._huge_numbers_as_string: 587 d = str(d) 588 589 class SomeService(Service): 590 @srpc(Integer, _returns=Integer) 591 def some_call(p): 592 print(p) 593 print(type(p)) 594 assert type(p) == long 595 return p 596 597 ctx = _dry_me([SomeService], {"some_call":[d]}) 598 599 s = self.loads(b''.join(ctx.out_string)) 600 d = {"some_callResponse": {"some_callResult": d}} 601 print(s) 602 print(d) 603 assert s == d 604 605 def test_time(self): 606 d = time(10, 20, 30).isoformat() 607 608 class SomeService(Service): 609 @srpc(Time, _returns=Time) 610 def some_call(p): 611 print(p) 612 print(type(p)) 613 assert type(p) == time 614 assert p.isoformat() == d 615 return p 616 617 ctx = _dry_me([SomeService], {"some_call":[d]}) 618 619 s = self.loads(b''.join(ctx.out_string)) 620 d = {"some_callResponse": {"some_callResult": d}} 621 print(s) 622 print(d) 623 assert s == d 624 625 def test_date(self): 626 vdt = datetime(2010, 9, 8) 627 d = vdt.date().isoformat() 628 629 class SomeService(Service): 630 @srpc(Date, _returns=Date) 631 def some_call(p): 632 print(p) 633 print(type(p)) 634 assert type(p) == date 635 assert p.isoformat() == d 636 return p 637 638 @srpc(_returns=Date) 639 def some_call_dt(): 640 return vdt 641 642 ctx = _dry_me([SomeService], {"some_call": [d]}) 643 s = self.loads(b''.join(ctx.out_string)) 644 rd = {"some_callResponse": {"some_callResult": d}} 645 print(s) 646 print(rd) 647 assert s == rd 648 649 ctx = _dry_me([SomeService], {"some_call_dt": []}) 650 s = self.loads(b''.join(ctx.out_string)) 651 rd = {"some_call_dtResponse": {"some_call_dtResult": d}} 652 print(s) 653 print(rd) 654 assert s == rd 655 656 def test_datetime(self): 657 d = datetime(2010, 9, 8, 7, 6, 5).isoformat() 658 659 class SomeService(Service): 660 @srpc(DateTime, _returns=DateTime(timezone=False)) 661 def some_call(p): 662 print(p) 663 print(type(p)) 664 assert type(p) == datetime 665 assert p.replace(tzinfo=None).isoformat() == d 666 return p 667 668 ctx = _dry_me([SomeService], {"some_call":[d]}, validator='soft') 669 670 s = self.loads(b''.join(ctx.out_string)) 671 d = {"some_callResponse": {"some_callResult": d}} 672 print(s) 673 print(d) 674 assert s == d 675 676 def test_datetime_tz(self): 677 d = datetime(2010, 9, 8, 7, 6, 5, tzinfo=pytz.utc).isoformat() 678 679 class SomeService(Service): 680 @srpc(DateTime, _returns=DateTime(ge=datetime(2010,1,1,tzinfo=pytz.utc))) 681 def some_call(p): 682 print(p) 683 print(type(p)) 684 assert type(p) == datetime 685 assert p.isoformat() == d 686 return p 687 688 ctx = _dry_me([SomeService], {"some_call":[d]}, validator='soft') 689 690 s = self.loads(b''.join(ctx.out_string)) 691 d = {"some_callResponse": {"some_callResult": d}} 692 print(s) 693 print(d) 694 assert s == d 695 696 def test_duration(self): 697 d = ProtocolBase().to_unicode(Duration, timedelta(0, 45)) 698 699 class SomeService(Service): 700 @srpc(Duration, _returns=Duration) 701 def some_call(p): 702 print(p) 703 print(type(p)) 704 assert type(p) == timedelta 705 return p 706 707 ctx = _dry_me([SomeService], {"some_call":[d]}) 708 709 s = self.loads(b''.join(ctx.out_string)) 710 d = {"some_callResponse": {"some_callResult": d}} 711 print(s) 712 print(d) 713 assert s == d 714 715 def test_boolean(self): 716 d = True 717 718 class SomeService(Service): 719 @srpc(Boolean, _returns=Boolean) 720 def some_call(p): 721 print(p) 722 print(type(p)) 723 assert type(p) == bool 724 return p 725 726 ctx = _dry_me([SomeService], {"some_call":[d]}) 727 728 s = self.loads(b''.join(ctx.out_string)) 729 d = {"some_callResponse": {"some_callResult": d}} 730 print(s) 731 print(d) 732 assert s == d 733 734 def test_uuid(self): 735 d = '7d2a6330-eb64-4900-8a10-38ebef415e9d' 736 737 class SomeService(Service): 738 @srpc(Uuid, _returns=Uuid) 739 def some_call(p): 740 print(p) 741 print(type(p)) 742 assert type(p) == uuid.UUID 743 return p 744 745 ctx = _dry_me([SomeService], {"some_call":[d]}) 746 747 s = self.loads(b''.join(ctx.out_string)) 748 d = {"some_callResponse": {"some_callResult": d}} 749 print(s) 750 print(d) 751 assert s == d 752 753 def test_point2d(self): 754 d = 'POINT(1 2)' 755 756 class SomeService(Service): 757 @srpc(Point, _returns=Point) 758 def some_call(p): 759 print(p) 760 print(type(p)) 761 assert isinstance(p, six.string_types) 762 return p 763 764 ctx = _dry_me([SomeService], {"some_call":[d]}) 765 766 s = self.loads(b''.join(ctx.out_string)) 767 d = {"some_callResponse": {"some_callResult": d}} 768 print(s) 769 print(d) 770 assert s == d 771 772 def test_point3d(self): 773 d = 'POINT(1 2 3)' 774 775 class SomeService(Service): 776 @srpc(Point, _returns=Point) 777 def some_call(p): 778 print(p) 779 print(type(p)) 780 assert isinstance(p, six.string_types) 781 return p 782 783 ctx = _dry_me([SomeService], {"some_call":[d]}) 784 785 s = self.loads(b''.join(ctx.out_string)) 786 d = {"some_callResponse": {"some_callResult": d}} 787 print(s) 788 print(d) 789 assert s == d 790 791 def test_line2d(self): 792 d = 'LINESTRING(1 2, 3 4)' 793 794 class SomeService(Service): 795 @srpc(Line, _returns=Line) 796 def some_call(p): 797 print(p) 798 print(type(p)) 799 assert isinstance(p, six.string_types) 800 return p 801 802 ctx = _dry_me([SomeService], {"some_call":[d]}) 803 804 s = self.loads(b''.join(ctx.out_string)) 805 d = {"some_callResponse": {"some_callResult": d}} 806 print(s) 807 print(d) 808 assert s == d 809 810 def test_line3d(self): 811 d = 'LINESTRING(1 2 3, 4 5 6)' 812 813 class SomeService(Service): 814 @srpc(Line, _returns=Line) 815 def some_call(p): 816 print(p) 817 print(type(p)) 818 assert isinstance(p, six.string_types) 819 return p 820 821 ctx = _dry_me([SomeService], {"some_call":[d]}) 822 823 s = self.loads(b''.join(ctx.out_string)) 824 d = {"some_callResponse": {"some_callResult": d}} 825 print(s) 826 print(d) 827 assert s == d 828 829 def test_polygon2d(self): 830 d = 'POLYGON((1 1, 1 2, 2 2, 2 1, 1 1))' 831 832 class SomeService(Service): 833 @srpc(Polygon(2), _returns=Polygon(2)) 834 def some_call(p): 835 print(p) 836 print(type(p)) 837 assert isinstance(p, six.string_types) 838 return p 839 840 ctx = _dry_me([SomeService], {"some_call":[d]}) 841 842 s = self.loads(b''.join(ctx.out_string)) 843 d = {"some_callResponse": {"some_callResult": d}} 844 print(s) 845 print(d) 846 assert s == d 847 848 def test_polygon3d(self): 849 d = 'POLYGON((1 1 0, 1 2 0, 2 2 0, 2 1 0, 1 1 0))' 850 851 class SomeService(Service): 852 @srpc(Polygon(3), _returns=Polygon(3)) 853 def some_call(p): 854 print(p) 855 print(type(p)) 856 assert isinstance(p, six.string_types) 857 return p 858 859 ctx = _dry_me([SomeService], {"some_call":[d]}) 860 861 s = self.loads(b''.join(ctx.out_string)) 862 d = {"some_callResponse": {"some_callResult": d}} 863 print(s) 864 print(d) 865 assert s == d 866 867 def test_multipoint2d(self): 868 d = 'MULTIPOINT ((10 40), (40 30), (20 20), (30 10))' 869 870 class SomeService(Service): 871 @srpc(MultiPoint(2), _returns=MultiPoint(2)) 872 def some_call(p): 873 print(p) 874 print(type(p)) 875 assert isinstance(p, six.string_types) 876 return p 877 878 ctx = _dry_me([SomeService], {"some_call":[d]}) 879 880 s = self.loads(b''.join(ctx.out_string)) 881 d = {"some_callResponse": {"some_callResult": d}} 882 print(s) 883 print(d) 884 assert s == d 885 886 def test_multipoint3d(self): 887 d = 'MULTIPOINT (10 40 30, 40 30 10,)' 888 889 class SomeService(Service): 890 @srpc(MultiPoint(3), _returns=MultiPoint(3)) 891 def some_call(p): 892 print(p) 893 print(type(p)) 894 assert isinstance(p, six.string_types) 895 return p 896 897 ctx = _dry_me([SomeService], {"some_call":[d]}) 898 899 s = self.loads(b''.join(ctx.out_string)) 900 d = {"some_callResponse": {"some_callResult": d}} 901 print(s) 902 print(d) 903 assert s == d 904 905 def test_multiline2d(self): 906 d = 'MULTILINESTRING ((10 10, 20 20, 10 40), (40 40, 30 30, 40 20, 30 10))' 907 908 class SomeService(Service): 909 @srpc(MultiLine(2), _returns=MultiLine(2)) 910 def some_call(p): 911 print(p) 912 print(type(p)) 913 assert isinstance(p, six.string_types) 914 return p 915 916 ctx = _dry_me([SomeService], {"some_call":[d]}) 917 918 s = self.loads(b''.join(ctx.out_string)) 919 d = {"some_callResponse": {"some_callResult": d}} 920 print(s) 921 print(d) 922 assert s == d 923 924 def test_multiline3d(self): 925 d = 'MULTILINESTRING ((10 10, 20 20, 10 40), (40 40, 30 30, 40 20, 30 10))' 926 927 class SomeService(Service): 928 @srpc(MultiLine(3), _returns=MultiLine(3)) 929 def some_call(p): 930 print(p) 931 print(type(p)) 932 assert isinstance(p, six.string_types) 933 return p 934 935 ctx = _dry_me([SomeService], {"some_call":[d]}) 936 937 s = self.loads(b''.join(ctx.out_string)) 938 d = {"some_callResponse": {"some_callResult": d}} 939 print(s) 940 print(d) 941 assert s == d 942 943 def test_multipolygon2d(self): 944 d = 'MULTIPOLYGON (((30 20, 10 40, 45 40, 30 20)),((15 5, 40 10, 10 20, 5 10, 15 5)))' 945 946 class SomeService(Service): 947 @srpc(MultiPolygon(2), _returns=MultiPolygon(2)) 948 def some_call(p): 949 print(p) 950 print(type(p)) 951 assert isinstance(p, six.string_types) 952 return p 953 954 ctx = _dry_me([SomeService], {"some_call":[d]}) 955 956 s = self.loads(b''.join(ctx.out_string)) 957 d = {"some_callResponse": {"some_callResult": d}} 958 print(s) 959 print(d) 960 assert s == d 961 962 def test_multipolygon3d(self): 963 d = 'MULTIPOLYGON (((40 40, 20 45, 45 30, 40 40)),' \ 964 '((20 35, 45 20, 30 5, 10 10, 10 30, 20 35),' \ 965 '(30 20, 20 25, 20 15, 30 20)))' 966 967 class SomeService(Service): 968 @srpc(MultiPolygon(3), _returns=MultiPolygon(3)) 969 def some_call(p): 970 print(p) 971 print(type(p)) 972 assert isinstance(p, six.string_types) 973 return p 974 975 ctx = _dry_me([SomeService], {"some_call":[d]}) 976 977 s = self.loads(b''.join(ctx.out_string)) 978 d = {"some_callResponse": {"some_callResult": d}} 979 print(s) 980 print(d) 981 assert s == d 982 983 def test_generator(self): 984 class SomeService(Service): 985 @srpc(_returns=Iterable(Integer)) 986 def some_call(): 987 return iter(range(1000)) 988 989 ctx = _dry_me([SomeService], {"some_call":[]}) 990 991 s = self.loads(b''.join(ctx.out_string)) 992 d = {"some_callResponse": {"some_callResult": tuple(range(1000))}} 993 print(s) 994 print(d) 995 assert s == d 996 997 def test_bytearray(self): 998 dbe = _DictDocumentChild.default_binary_encoding 999 beh = binary_encoding_handlers[dbe] 1000 1001 data = bytes(bytearray(range(0xff))) 1002 encoded_data = beh([data]) 1003 if _DictDocumentChild.text_based: 1004 encoded_data = encoded_data.decode('latin1') 1005 1006 class SomeService(Service): 1007 @srpc(ByteArray, _returns=ByteArray) 1008 def some_call(ba): 1009 print(ba) 1010 print(type(ba)) 1011 assert isinstance(ba, tuple) 1012 assert ba == (data,) 1013 return ba 1014 1015 ctx = _dry_me([SomeService], {"some_call": [encoded_data]}) 1016 1017 s = self.loads(b''.join(ctx.out_string)) 1018 d = {"some_callResponse": {"some_callResult": encoded_data}} 1019 1020 print(repr(s)) 1021 print(repr(d)) 1022 print(repr(encoded_data)) 1023 assert s == d 1024 1025 def test_file_data(self): 1026 # the only difference with the bytearray test is/are the types 1027 # inside @srpc 1028 dbe = _DictDocumentChild.default_binary_encoding 1029 beh = binary_encoding_handlers[dbe] 1030 1031 data = bytes(bytearray(range(0xff))) 1032 encoded_data = beh([data]) 1033 if _DictDocumentChild.text_based: 1034 encoded_data = encoded_data.decode('latin1') 1035 1036 class SomeService(Service): 1037 @srpc(File, _returns=File) 1038 def some_call(p): 1039 print(p) 1040 print(type(p)) 1041 assert isinstance(p, File.Value) 1042 assert p.data == (data,) 1043 return p.data 1044 1045 # we put the encoded data in the list of arguments. 1046 ctx = _dry_me([SomeService], {"some_call": [encoded_data]}) 1047 1048 s = self.loads(b''.join(ctx.out_string)) 1049 d = {"some_callResponse": {"some_callResult": encoded_data}} 1050 1051 print(s) 1052 print(d) 1053 print(repr(encoded_data)) 1054 assert s == d 1055 1056 def test_file_value(self): 1057 dbe = _DictDocumentChild.default_binary_encoding 1058 beh = binary_encoding_handlers[dbe] 1059 1060 # Prepare data 1061 v = File.Value( 1062 name='some_file.bin', 1063 type='application/octet-stream', 1064 ) 1065 file_data = bytes(bytearray(range(0xff))) 1066 v.data = (file_data,) 1067 beh([file_data]) 1068 if _DictDocumentChild.text_based: 1069 test_data = beh(v.data).decode('latin1') 1070 else: 1071 test_data = beh(v.data) 1072 1073 print(repr(v.data)) 1074 1075 class SomeService(Service): 1076 @srpc(File, _returns=File) 1077 def some_call(p): 1078 print(p) 1079 print(type(p)) 1080 assert isinstance(p, File.Value) 1081 assert p.data == (file_data,) 1082 assert p.type == v.type 1083 assert p.name == v.name 1084 return p 1085 1086 d = get_object_as_dict(v, File, protocol=_DictDocumentChild, 1087 ignore_wrappers=False) 1088 ctx = _dry_me([SomeService], {"some_call": {'p': d}}) 1089 s = b''.join(ctx.out_string) 1090 d = self.dumps({"some_callResponse": {"some_callResult": { 1091 'name': v.name, 1092 'type': v.type, 1093 'data': test_data, 1094 }}}) 1095 1096 print(self.loads(s)) 1097 print(self.loads(d)) 1098 print(v) 1099 assert self.loads(s) == self.loads(d) 1100 1101 def test_validation_frequency(self): 1102 class SomeService(Service): 1103 @srpc(ByteArray(min_occurs=1), _returns=ByteArray) 1104 def some_call(p): 1105 pass 1106 1107 try: 1108 _dry_me([SomeService], {"some_call": []}, validator='soft') 1109 except ValidationError: 1110 pass 1111 else: 1112 raise Exception("must raise ValidationError") 1113 1114 def test_validation_nullable(self): 1115 class SomeService(Service): 1116 @srpc(ByteArray(nullable=False), _returns=ByteArray) 1117 def some_call(p): 1118 pass 1119 1120 try: 1121 _dry_me([SomeService], {"some_call": [None]}, 1122 validator='soft') 1123 except ValidationError: 1124 pass 1125 1126 else: 1127 raise Exception("must raise ValidationError") 1128 1129 def test_validation_string_pattern(self): 1130 class SomeService(Service): 1131 @srpc(Uuid) 1132 def some_call(p): 1133 pass 1134 1135 try: 1136 _dry_me([SomeService], {"some_call": ["duduk"]}, 1137 validator='soft') 1138 except ValidationError as e: 1139 print(e) 1140 pass 1141 1142 else: 1143 raise Exception("must raise ValidationError") 1144 1145 def test_validation_integer_range(self): 1146 class SomeService(Service): 1147 @srpc(Integer(ge=0, le=5)) 1148 def some_call(p): 1149 pass 1150 1151 try: 1152 _dry_me([SomeService], {"some_call": [10]}, 1153 validator='soft') 1154 except ValidationError: 1155 pass 1156 1157 else: 1158 raise Exception("must raise ValidationError") 1159 1160 def test_validation_integer_type(self): 1161 class SomeService(Service): 1162 @srpc(Integer8) 1163 def some_call(p): 1164 pass 1165 1166 try: 1167 _dry_me([SomeService], {"some_call": [-129]}, 1168 validator='soft') 1169 except ValidationError: 1170 pass 1171 1172 else: 1173 raise Exception("must raise ValidationError") 1174 1175 def test_validation_integer_type_2(self): 1176 class SomeService(Service): 1177 @srpc(Integer8) 1178 def some_call(p): 1179 pass 1180 1181 try: 1182 _dry_me([SomeService], {"some_call": [1.2]}, validator='soft') 1183 1184 except ValidationError: 1185 pass 1186 1187 else: 1188 raise Exception("must raise ValidationError") 1189 1190 def test_not_wrapped(self): 1191 class SomeInnerClass(ComplexModel): 1192 d = date 1193 dt = datetime 1194 1195 class SomeClass(ComplexModel): 1196 a = int 1197 b = Unicode 1198 c = SomeInnerClass.customize(not_wrapped=True) 1199 1200 class SomeService(Service): 1201 @srpc(SomeClass.customize(not_wrapped=True), 1202 _returns=SomeClass.customize(not_wrapped=True)) 1203 def some_call(p): 1204 assert p.a == 1 1205 assert p.b == 's' 1206 assert p.c.d == date(2018, 11, 22) 1207 return p 1208 1209 inner = {"a": 1, "b": "s", "c": {"d": '2018-11-22'}} 1210 doc = {"some_call": [inner]} 1211 ctx = _dry_me([SomeService], doc, validator='soft') 1212 1213 print(ctx.out_document) 1214 1215 d = convert_dict({"some_callResponse": {"some_callResult": inner}}) 1216 self.assertEquals(ctx.out_document[0], d) 1217 1218 def test_validation_freq_parent(self): 1219 class C(ComplexModel): 1220 i = Integer(min_occurs=1) 1221 s = Unicode 1222 1223 class SomeService(Service): 1224 @srpc(C) 1225 def some_call(p): 1226 pass 1227 1228 try: 1229 # must raise validation error for missing i 1230 _dry_me([SomeService], {"some_call": {'p': {'C': {'s': 'a'}}}}, 1231 validator='soft') 1232 except ValidationError as e: 1233 logger.exception(e) 1234 pass 1235 except BaseException as e: 1236 logger.exception(e) 1237 pass 1238 else: 1239 raise Exception("must raise ValidationError") 1240 1241 # must not raise anything for missing p because C has min_occurs=0 1242 _dry_me([SomeService], {"some_call": {}}, validator='soft') 1243 1244 def test_inheritance(self): 1245 class P(ComplexModel): 1246 identifier = Uuid 1247 signature = Unicode 1248 1249 class C(P): 1250 foo = Unicode 1251 bar = Uuid 1252 1253 class SomeService(Service): 1254 @rpc(_returns=C) 1255 def some_call(ctx): 1256 result = C() 1257 result.identifier = uuid.UUID(int=0) 1258 result.signature = 'yyyyyyyyyyy' 1259 result.foo = 'zzzzzz' 1260 result.bar = uuid.UUID(int=1) 1261 return result 1262 1263 ctx = _dry_me([SomeService], {"some_call": []}) 1264 1265 s = self.loads(b''.join(ctx.out_string)) 1266 d = {"some_callResponse": {"some_callResult": {"C": { 1267 'identifier': '00000000-0000-0000-0000-000000000000', 1268 'bar': '00000000-0000-0000-0000-000000000001', 1269 'foo': 'zzzzzz', 1270 'signature': 'yyyyyyyyyyy' 1271 }}}} 1272 1273 assert s == d 1274 1275 def test_exclude(self): 1276 class C(ComplexModel): 1277 s1 = Unicode(exc=True) 1278 s2 = Unicode 1279 1280 class SomeService(Service): 1281 @srpc(C, _returns=C) 1282 def some_call(sc): 1283 assert sc.s1 is None, "sc={}".format(sc) 1284 assert sc.s2 == "s2" 1285 return C(s1="s1", s2="s2") 1286 1287 doc = [{"C": {"s1": "s1","s2": "s2"}}] 1288 ctx = _dry_me([SomeService], {"some_call": doc}) 1289 1290 self.assertEquals(ctx.out_document[0], convert_dict( 1291 {'some_callResponse': {'some_callResult': {'C': {'s2': 's2'}}}}) 1292 ) 1293 1294 def test_polymorphic_deserialization(self): 1295 class P(ComplexModel): 1296 sig = Unicode 1297 1298 class C(P): 1299 foo = Unicode 1300 1301 class D(P): 1302 bar = Integer 1303 1304 class SomeService(Service): 1305 @rpc(P, _returns=Unicode) 1306 def typeof(ctx, p): 1307 return type(p).__name__ 1308 1309 ctx = _dry_me([SomeService], 1310 {"typeof": [{'C':{'sig':'a', 'foo': 'f'}}]}, 1311 polymorphic=True) 1312 1313 s = self.loads(b''.join(ctx.out_string)) 1314 d = {"typeofResponse": {"typeofResult": 'C'}} 1315 1316 print(s) 1317 print(d) 1318 assert s == d 1319 1320 ctx = _dry_me([SomeService], 1321 {"typeof": [{'D':{'sig':'b', 'bar': 5}}]}, 1322 polymorphic=True) 1323 1324 s = self.loads(b''.join(ctx.out_string)) 1325 d = {"typeofResponse": {"typeofResult": 'D'}} 1326 1327 print(s) 1328 print(d) 1329 assert s == d 1330 1331 def test_default(self): 1332 class SomeComplexModel(ComplexModel): 1333 _type_info = [ 1334 ('a', Unicode), 1335 ('b', Unicode(default='default')), 1336 ] 1337 1338 class SomeService(Service): 1339 @srpc(SomeComplexModel) 1340 def some_method(s): 1341 pass 1342 1343 ctx = _dry_me([SomeService], 1344 {"some_method": [{"s": {"a": "x", "b": None}}]}, 1345 polymorphic=True) 1346 1347 assert ctx.in_object.s.b == None 1348 assert ctx.in_error is None 1349 1350 ctx = _dry_me([SomeService], {"some_method": {"s": {"a": "x"}}}, 1351 polymorphic=True) 1352 1353 assert ctx.in_object.s.b == 'default' 1354 assert ctx.in_error is None 1355 1356 def test_nillable_default(self): 1357 class SomeComplexModel(ComplexModel): 1358 _type_info = [ 1359 ('a', Unicode), 1360 ('b', Unicode(min_occurs=1, default='default', nillable=True)), 1361 ] 1362 1363 class SomeService(Service): 1364 @srpc(SomeComplexModel) 1365 def some_method(s): 1366 pass 1367 1368 ctx = _dry_me([SomeService], 1369 {"some_method": [{"s": {"a": "x", "b": None}}]}, 1370 polymorphic=True, validator='soft') 1371 1372 assert ctx.in_object.s.b == None 1373 assert ctx.in_error is None 1374 1375 ctx = _dry_me([SomeService], {"some_method": {"s": {"a": "x"}}}, 1376 polymorphic=True) 1377 1378 assert ctx.in_object.s.b == 'default' 1379 assert ctx.in_error is None 1380 1381 return Test 1382
