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('RwProjectPersonDbYang', '1.0')
  38 gi.require_version('RwYang', '1.0')
  39
  40 from gi.repository import (
  41         RwpersonDbYang,
  42         RwProjectPersonDbYang,
  43         RwYang,
  44         )
  45
  46 from rift.package.convert import SerializationError
  47
  48
  49 class TestSerializer(unittest.TestCase):
  50     def setUp(self):
  51         self._serializer = ProtoMessageSerializer(
  52                 RwpersonDbYang,
  53                 RwpersonDbYang.Person,
  54                 RwProjectPersonDbYang,
  55                 RwProjectPersonDbYang.YangData_RwProject_Project_Person,
  56                 )
  57
  58         self._sample_person = RwpersonDbYang.Person(name="Fred")
  59         self._project_person = RwProjectPersonDbYang.YangData_RwProject_Project_Person(name="Fred")
  60         self._model = RwYang.model_create_libyang()
  61         self._model.load_schema_ypbc(RwpersonDbYang.get_schema())
  62
  63     def test_from_xml_file(self):
  64         sample_person_xml = self._sample_person.to_xml_v2(self._model)
  65         with io.StringIO(sample_person_xml) as file_hdl:
  66             person = self._serializer.from_file_hdl(file_hdl, ".xml")
  67             self.assertEqual(person, self._sample_person)
  68
  69     def test_from_yaml_file(self):
  70         sample_person_yaml = self._sample_person.to_yaml(self._model)
  71         with io.StringIO(sample_person_yaml) as file_hdl:
  72
  73             person = self._serializer.from_file_hdl(file_hdl, ".yml")
  74             self.assertEqual(person, self._project_person)
  75
  76     def test_from_json_file(self):
  77         sample_person_json = self._sample_person.to_json(self._model)
  78         with io.StringIO(sample_person_json) as file_hdl:
  79
  80             person = self._serializer.from_file_hdl(file_hdl, ".json")
  81             self.assertEqual(person, self._project_person)
  82
  83     def test_unknown_file_extension(self):
  84         with io.StringIO("asdf") as file_hdl:
  85             with self.assertRaises(UnknownExtensionError):
  86                 self._serializer.from_file_hdl(file_hdl, ".foo")
  87
  88     def test_raises_serialization_error(self):
  89         with io.StringIO('</foo>') as file_hdl:
  90             with self.assertRaises(SerializationError):
  91                 person = self._serializer.from_file_hdl(file_hdl, ".json")
  92                 print(person)
  93
  94     def test_to_json_string(self):
  95         json_str = self._serializer.to_json_string(self._sample_person)
  96
  97         person = RwpersonDbYang.Person.from_json(self._model, json_str)
  98         self.assertEqual(person, self._sample_person)
  99
 100     def test_to_json_string_invalid_type(self):
 101         with self.assertRaises(SerializationError):
 102             self._serializer.to_json_string(RwpersonDbYang.FlatPerson(name="bob"))
 103
 104
 105 def main(argv=sys.argv[1:]):
 106     logging.basicConfig(format='TEST %(message)s')
 107
 108     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 109     parser = argparse.ArgumentParser()
 110     parser.add_argument('-v', '--verbose', action='store_true')
 111     parser.add_argument('-n', '--no-runner', action='store_true')
 112
 113     args, unknown = parser.parse_known_args(argv)
 114     if args.no_runner:
 115         runner = None
 116
 117     # Set the global logging level
 118     logging.getLogger().setLevel(logging.DEBUG if args.verbose else logging.ERROR)
 119
 120     # The unittest framework requires a program name, so use the name of this
 121     # file instead (we do not want to have to pass a fake program name to main
 122     # when this is called from the interpreter).
 123     unittest.main(argv=[__file__] + unknown + ["-v"], testRunner=runner)
 124
 125 if __name__ == '__main__':
 126     main()