rwlaunchpad/plugins/rwlaunchpadtasklet/test/utest_serializer.py

   1 #!/usr/bin/env python3
   2
   3 #
   4 #   Copyright 2016 RIFT.IO Inc
   5 #
   6 #   Licensed under the Apache License, Version 2.0 (the "License");
   7 #   you may not use this file except in compliance with the License.
   8 #   You may obtain a copy of the License at
   9 #
  10 #       http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 #   Unless required by applicable law or agreed to in writing, software
  13 #   distributed under the License is distributed on an "AS IS" BASIS,
  14 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 #   See the License for the specific language governing permissions and
  16 #   limitations under the License.
  17 #
  18
  19
  20 import argparse
  21 import logging
  22 import io
  23 import os
  24 import sys
  25 import tempfile
  26 import unittest
  27 import xmlrunner
  28
  29 from rift.package.convert import (
  30         ProtoMessageSerializer,
  31         UnknownExtensionError,
  32         SerializationError,
  33         )
  34
  35 import gi
  36 gi.require_version('RwpersonDbYang', '1.0')
  37 gi.require_version('RwYang', '1.0')
  38
  39 from gi.repository import (
  40         RwpersonDbYang,
  41         RwYang,
  42         )
  43
  44 class TestSerializer(unittest.TestCase):
  45     def setUp(self):
  46         self._serializer = ProtoMessageSerializer(
  47                 RwpersonDbYang,
  48                 RwpersonDbYang.Person
  49                 )
  50
  51         self._sample_person = RwpersonDbYang.Person(name="Fred")
  52         self._model = RwYang.model_create_libncx()
  53         self._model.load_schema_ypbc(RwpersonDbYang.get_schema())
  54
  55     def test_from_xml_file(self):
  56         sample_person_xml = self._sample_person.to_xml_v2(self._model)
  57         with io.StringIO(sample_person_xml) as file_hdl:
  58             person = self._serializer.from_file_hdl(file_hdl, ".xml")
  59             self.assertEqual(person, self._sample_person)
  60
  61     def test_from_yaml_file(self):
  62         sample_person_yaml = self._sample_person.to_yaml(self._model)
  63         with io.StringIO(sample_person_yaml) as file_hdl:
  64
  65             person = self._serializer.from_file_hdl(file_hdl, ".yml")
  66             self.assertEqual(person, self._sample_person)
  67
  68     def test_from_json_file(self):
  69         sample_person_json = self._sample_person.to_json(self._model)
  70         with io.StringIO(sample_person_json) as file_hdl:
  71
  72             person = self._serializer.from_file_hdl(file_hdl, ".json")
  73             self.assertEqual(person, self._sample_person)
  74
  75     def test_unknown_file_extension(self):
  76         with io.StringIO("asdf") as file_hdl:
  77             with self.assertRaises(UnknownExtensionError):
  78                 self._serializer.from_file_hdl(file_hdl, ".foo")
  79
  80     def test_raises_serialization_error(self):
  81         with io.StringIO('</foo>') as file_hdl:
  82             with self.assertRaises(SerializationError):
  83                 person = self._serializer.from_file_hdl(file_hdl, ".json")
  84                 print(person)
  85
  86     def test_to_json_string(self):
  87         json_str = self._serializer.to_json_string(self._sample_person)
  88
  89         person = RwpersonDbYang.Person.from_json(self._model, json_str)
  90         self.assertEqual(person, self._sample_person)
  91
  92     def test_to_json_string_invalid_type(self):
  93         with self.assertRaises(TypeError):
  94             self._serializer.to_json_string(RwpersonDbYang.FlatPerson(name="bob"))
  95
  96
  97 def main(argv=sys.argv[1:]):
  98     logging.basicConfig(format='TEST %(message)s')
  99
 100     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 101     parser = argparse.ArgumentParser()
 102     parser.add_argument('-v', '--verbose', action='store_true')
 103     parser.add_argument('-n', '--no-runner', action='store_true')
 104
 105     args, unknown = parser.parse_known_args(argv)
 106     if args.no_runner:
 107         runner = None
 108
 109     # Set the global logging level
 110     logging.getLogger().setLevel(logging.DEBUG if args.verbose else logging.ERROR)
 111
 112     # The unittest framework requires a program name, so use the name of this
 113     # file instead (we do not want to have to pass a fake program name to main
 114     # when this is called from the interpreter).
 115     unittest.main(argv=[__file__] + unknown + ["-v"], testRunner=runner)
 116
 117 if __name__ == '__main__':
 118     main()