1 #!/usr/bin/python2.6

     2 #

     3 # CDDL HEADER START

     4 #

     5 # The contents of this file are subject to the terms of the

     6 # Common Development and Distribution License (the "License").

     7 # You may not use this file except in compliance with the License.

     8 #

     9 # You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

    10 # or http://www.opensolaris.org/os/licensing.

    11 # See the License for the specific language governing permissions

    12 # and limitations under the License.

    13 #

    14 # When distributing Covered Code, include this CDDL HEADER in each

    15 # file and include the License file at usr/src/OPENSOLARIS.LICENSE.

    16 # If applicable, add the following below this CDDL HEADER, with the

    17 # fields enclosed by brackets "[]" replaced with your own identifying

    18 # information: Portions Copyright [yyyy] [name of copyright owner]

    19 #

    20 # CDDL HEADER END

    21 #

    22 # Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.

    23 #

    24 

    25 '''

    26 To run these tests, see the instructions in usr/src/tools/tests/README.

    27 Remember that since the proto area is used for the PYTHONPATH, the gate

    28 must be rebuilt for these tests to pick up any changes in the tested code.

    29 

    30 '''

    31 

    32 import gettext

    33 import tempfile

    34 import unittest

    35 import set_criteria as set_criteria

    36 import osol_install.auto_install.AI_database as AIdb

    37 

    38 

    39 gettext.install("ai-test")

    40 

    41 class MockDataBase(object):

    42     '''Class for mock database '''

    43     def __init__(self):

    44         self.queue  = MockQueue()

    45 

    46     def getQueue(self):

    47         return self.queue

    48 

    49 class MockQueue(object):

    50     '''Class for mock database '''

    51     def __init__(self):

    52         self.criteria = None

    53 

    54     def put(self, query):

    55         return

    56 

    57 class MockQuery(object):

    58     '''Class for mock query '''

    59     def __init__(self):

    60         self.query = None

    61 

    62     def __call__(self, query, commit=False):

    63         self.query = query

    64         return self

    65 

    66     def waitAns(self):

    67         return

    68 

    69     def getResponse(self):

    70         return

    71 

    72 class MockGetCriteria(object):

    73     '''Class for mock getCriteria '''

    74     def __init__(self):

    75         self.crit_stripped = ["arch", "mem", "ipv4", "mac"]

    76         self.crit_unstripped = ["MINmem", "MINipv4", "MINmac",

    77                                 "MAXmem", "MAXipv4", "MAXmac", "arch"]

    78 

    79     def __call__(self, queue, onlyUsed=False, strip=False):

    80         if strip:

    81             return self.crit_stripped

    82         else:

    83             return self.crit_unstripped

    84 

    85 class MockisRangeCriteria(object):

    86     '''Class for mock isRangeCriteria '''

    87     def __init__(self):

    88         self.range = ["mem", "ipv4", "mac"]

    89 

    90     def __call__(self, queue, crit):

    91         if crit in self.range:

    92             return True

    93         return False

    94 

    95 class MockgetManNames(object):

    96     '''Class for mock getManNames '''

    97     def __init__(self):

    98         self.man_names = ["fakeman"]

    99 

   100     def __call__(self, queue):

   101         return self.man_names

   102 

   103 class MockDataFiles(object):

   104     '''Class for mock DataFiles'''

   105     def __init__(self):

   106         self.criteria = None

   107         self.database = MockDataBase()

   108 

   109 class SetCriteria(unittest.TestCase):

   110     '''Tests for set_criteria'''

   111 

   112     def setUp(self):

   113         '''unit test set up

   114 

   115         '''

   116         self.aidb_DBrequest = AIdb.DBrequest

   117         self.aidb_isRangeCriteria = AIdb.isRangeCriteria

   118         self.aidb_getCriteria = AIdb.getCriteria

   119         self.mockquery = MockQuery()

   120         self.mockgetCriteria = MockGetCriteria()

   121         self.mockisRangeCriteria = MockisRangeCriteria()

   122         AIdb.DBrequest = self.mockquery

   123         AIdb.getCriteria = self.mockgetCriteria

   124         AIdb.isRangeCriteria = self.mockisRangeCriteria

   125         self.files = MockDataFiles()

   126 

   127     def tearDown(self):

   128         '''unit test tear down

   129         Functions originally saved in setUp are restored to their

   130         original values.

   131         '''

   132         AIdb.DBrequest = self.aidb_DBrequest

   133         AIdb.getCriteria = self.aidb_getCriteria

   134         AIdb.isRangeCriteria = self.aidb_isRangeCriteria

   135 

   136     def test_unbounded_min(self):

   137         '''Ensure set_criteria min query constructed properly '''

   138         criteria = {"arch": "i86pc", "mem": ["unbounded", 4096]}

   139         criteria.setdefault("ipv4")

   140         criteria.setdefault("mac")

   141         set_criteria.set_criteria(criteria, "myxml", self.files.database)

   142         expect_query = "UPDATE manifests SET arch='i86pc',MINmem=NULL," + \

   143                        "MAXmem='4096',MINipv4=NULL,MAXipv4=NULL,MINmac=NULL," +\

   144                        "MAXmac=NULL WHERE name='myxml'"

   145         self.assertEquals(expect_query, self.mockquery.query)

   146 

   147     def test_unbounded_max(self):

   148         '''Ensure set_criteria max query constructed properly '''

   149         criteria = {"arch": "i86pc", "mem": [1024, "unbounded"]}

   150         criteria.setdefault("ipv4")

   151         criteria.setdefault("mac")

   152         set_criteria.set_criteria(criteria, "myxml", self.files.database)

   153         expect_query = "UPDATE manifests SET arch='i86pc',MINmem='1024'," + \

   154                        "MAXmem=NULL,MINipv4=NULL,MAXipv4=NULL,MINmac=NULL," + \

   155                        "MAXmac=NULL WHERE name='myxml'"

   156         self.assertEquals(expect_query, self.mockquery.query)

   157 

   158     def test_range(self):

   159         '''Ensure set_criteria max query constructed properly '''

   160         criteria = {"arch": "i86pc", "ipv4": ["10.0.30.100", "10.0.50.400"]}

   161         criteria.setdefault("mac")

   162         criteria.setdefault("mem")

   163         set_criteria.set_criteria(criteria, "myxml", self.files.database)

   164         expect_query = "UPDATE manifests SET arch='i86pc',MINmem=NULL," + \

   165                        "MAXmem=NULL,MINipv4='10.0.30.100'," + \

   166                        "MAXipv4='10.0.50.400',MINmac=NULL,MAXmac=NULL " + \

   167                        "WHERE name='myxml'"

   168         self.assertEquals(expect_query, self.mockquery.query)

   169 

   170     def test_append_unbounded_min(self):

   171         '''Ensure set_criteria append min query constructed properly '''

   172         criteria = {"arch": "i86pc", "mem": ["unbounded", 4096]}

   173         criteria.setdefault("ipv4")

   174         criteria.setdefault("mac")

   175         set_criteria.set_criteria(criteria, "myxml", self.files.database,

   176                                   append=True)

   177         expect_query = "UPDATE manifests SET arch='i86pc',MINmem=NULL," \

   178                        "MAXmem='4096' WHERE name='myxml'"

   179         self.assertEquals(expect_query, self.mockquery.query)

   180 

   181     def test_append_unbounded_max(self):

   182         '''Ensure set_criteria append max query constructed properly '''

   183         criteria = {"arch": "i86pc", "mem": [2048, "unbounded"]}

   184         criteria.setdefault("ipv4")

   185         criteria.setdefault("mac")

   186         set_criteria.set_criteria(criteria, "myxml", self.files.database,

   187                                   append=True)

   188         expect_query = "UPDATE manifests SET arch='i86pc',MINmem='2048'," \

   189                        "MAXmem=NULL WHERE name='myxml'"

   190         self.assertEquals(expect_query, self.mockquery.query)

   191 

   192     def test_append_range(self):

   193         '''Ensure set_criteria append range query constructed properly '''

   194         criteria = {"arch": "i86pc", "ipv4": ["10.0.10.10", "10.0.10.300"]}

   195         criteria.setdefault("mem")

   196         criteria.setdefault("mac")

   197         set_criteria.set_criteria(criteria, "myxml", self.files.database,

   198                                   append=True)

   199         expect_query = "UPDATE manifests SET arch='i86pc',MINipv4=" + \

   200                        "'10.0.10.10',MAXipv4='10.0.10.300' WHERE name='myxml'"

   201         self.assertEquals(expect_query, self.mockquery.query)

   202 

   203 class CheckPublishedManifest(unittest.TestCase):

   204     '''Tests for check_published_manifest'''

   205 

   206     def setUp(self):

   207         '''unit test set up

   208 

   209         '''

   210         self.aidb_DBrequest = AIdb.DBrequest

   211         self.mockquery = MockQuery()

   212         AIdb.DBrequest = self.mockquery

   213         self.aidb_getManNames = AIdb.getManNames

   214         self.mockgetManNames = MockgetManNames()

   215         AIdb.getManNames = self.mockgetManNames

   216         self.files = MockDataFiles()

   217 

   218     def tearDown(self):

   219         '''unit test tear down

   220         Functions originally saved in setUp are restored to their

   221         original values.

   222         '''

   223         AIdb.DBrequest = self.aidb_DBrequest

   224         AIdb.getManNames = self.aidb_getManNames

   225 

   226     def test_no_such_name(self):

   227         '''Check no such manifest name caught '''

   228         self.assertFalse(set_criteria.check_published_manifest("/tmp",

   229                          self.files.database, "no_such_manifest"))

   230 

   231     def test_manifest_not_published(self):

   232         '''Check manifest not in published area caught'''

   233         self.assertFalse(set_criteria.check_published_manifest("/tmp",

   234                          self.files.database,

   235                          self.mockgetManNames.man_names[0])) 

   236 

   237 class ParseOptions(unittest.TestCase):

   238     '''Tests for parse_options. Some tests correctly output usage msg'''

   239 

   240     def test_parse_no_options(self):

   241         '''Ensure no options caught'''

   242         self.assertRaises(SystemExit, set_criteria.parse_options, []) 

   243         myargs = ["mysvc"] 

   244         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   245         myargs = ["manifest"] 

   246         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   247         myargs = ["mysvc", "manifest"] 

   248         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   249 

   250     def test_parse_invalid_options(self):

   251         '''Ensure invalid option flagged'''

   252         myargs = ["-n", "mysvc", "-m", "manifest", "-u"] 

   253         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   254 

   255     def test_parse_options_novalue(self):

   256         '''Ensure options with missing value caught'''

   257         myargs = ["-n", "mysvc", "-m", "manifest", "-c"] 

   258         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   259         myargs = ["-n", "mysvc", "-m", "manifest", "-C"] 

   260         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   261         myargs = ["-n", "mysvc", "-m", "manifest", "-a"] 

   262         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   263         myargs = ["-n", "-m", "manifest"]

   264         self.assertRaises(SystemExit, set_criteria.parse_options, myargs)

   265         myargs = ["-n", "mysvc", "-m"]

   266         self.assertRaises(SystemExit, set_criteria.parse_options, myargs)

   267 

   268     def test_parse_mutually_exclusive(self):

   269         '''Ensure mutually exclusive options caught'''

   270         # Ensure -C and -a caught

   271         myargs = ["-n", "mysvc", "-m", "manifest", "-a", "arch=i86pc",

   272                   "-C", tempfile.mktemp()] 

   273         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   274 

   275         # Ensure -C and -c caught

   276         myargs = ["-n", "mysvc", "-m", "manifest", "-c", "arch=i86pc", "-C",

   277                   tempfile.mktemp()] 

   278         self.assertRaises(SystemExit, set_criteria.parse_options, myargs) 

   279 

   280     def test_parse_valid_options(self):

   281         '''Ensure valid options parse successfully'''

   282         myargs = ["-n", "mysvc", "-m", "manifest", "-a", "arch=i86pc"]

   283         exp_options = {'criteria_a': ['arch=i86pc'], 'service_name': 'mysvc',

   284                        'manifest_name': 'manifest',

   285                        'criteria_file': None, 'criteria_c': []}

   286         options = set_criteria.parse_options(myargs)

   287         self.assertEquals(exp_options['criteria_a'], options.criteria_a) 

   288         self.assertEquals(exp_options['criteria_c'], options.criteria_c) 

   289         self.assertEquals(exp_options['service_name'], options.service_name)

   290         self.assertEquals(exp_options['manifest_name'], options.manifest_name) 

   291         self.assertEquals(exp_options['criteria_file'], options.criteria_file)

   292 

   293         myargs = ["-n", "mysvc", "-m", "manifest", "-c", "arch=i86pc"]

   294         exp_options = {'criteria_a': [],

   295                        'service_name': 'mysvc', 'manifest_name': 'manifest',

   296                        'criteria_file': None, 'criteria_c': ['arch=i86pc']}

   297         options = set_criteria.parse_options(myargs)

   298         self.assertEquals(exp_options['criteria_a'], options.criteria_a) 

   299         self.assertEquals(exp_options['criteria_c'], options.criteria_c) 

   300         self.assertEquals(exp_options['service_name'], options.service_name)

   301         self.assertEquals(exp_options['manifest_name'], options.manifest_name) 

   302         self.assertEquals(exp_options['criteria_file'], options.criteria_file)

   303 

   304         tempname = tempfile.mktemp()

   305         myargs = ["-n", "mysvc", "-m", "manifest", "-C", tempname]

   306         exp_options = {'criteria_a': [], 'service_name': 'mysvc',

   307                        'manifest_name': 'manifest', 'criteria_file': tempname,

   308                        'criteria_c': []}

   309         options = set_criteria.parse_options(myargs)

   310         self.assertEquals(exp_options['criteria_a'], options.criteria_a) 

   311         self.assertEquals(exp_options['criteria_c'], options.criteria_c) 

   312         self.assertEquals(exp_options['service_name'], options.service_name)

   313         self.assertEquals(exp_options['manifest_name'], options.manifest_name) 

   314         self.assertEquals(exp_options['criteria_file'], options.criteria_file)

   315 

   316 if __name__ == '__main__':

   317     unittest.main()