blob: 314aca6988f80b9b22543d1c548a62c0415f11f0 [file] [log] [blame]
#!/usr/bin/python
# Copyright 2014 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# TR-069 has mandatory attribute names that don't comply with policy
# pylint:disable=invalid-name
# pylint:disable=unused-argument
#
"""Tests for experiment.py."""
import os.path
import shutil
import tempfile
import google3
import core
import experiment
from wvtest import unittest
@experiment.Experiment
def TestExp1(h):
return [('TestParam', 1)]
@experiment.Experiment
def TestExp2(h):
yield ('SubObj.TestParam', h.obj.TestParam)
@experiment.Experiment
def TestExpOverlap(h):
return [('TestParam', 2),
('SubObj.SubObj.TestParam', 'Hello')]
class TestObject(core.Exporter):
def __init__(self):
core.Exporter.__init__(self)
self.Export(params=['TestParam'],
objects=['SubObj'])
self.TestParam = 5
self.SubObj = None
class TerminalObject(core.Exporter):
def __init__(self):
core.Exporter.__init__(self)
self.Export(params=['TestParam'])
self.TestParam = 5
class ExperimentTest(unittest.TestCase):
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
self.old_REGDIR = experiment.REGDIR
self.old_ACTIVEDIR = experiment.ACTIVEDIR
experiment.REGDIR = os.path.join(self.tmpdir, 'reg')
experiment.ACTIVEDIR = os.path.join(self.tmpdir, 'act')
os.mkdir(experiment.REGDIR)
os.mkdir(experiment.ACTIVEDIR)
def tearDown(self):
experiment.REGDIR = self.old_REGDIR
experiment.ACTIVEDIR = self.old_ACTIVEDIR
shutil.rmtree(self.tmpdir)
def testExperiments(self):
root = TestObject()
root.SubObj = TestObject()
root.SubObj.SubObj = TerminalObject()
eh = experiment.ExperimentHandle(root)
exps = experiment.Experiments(eh)
eh.root_experiments = exps
self.assertEqual(exps.Active, '')
self.assertTrue('TestExp1' in exps.Available.split(','))
self.assertTrue('TestExp2' in exps.Available.split(','))
self.assertTrue('TestExpOverlap' in exps.Available.split(','))
def Vals():
return (root.TestParam,
root.SubObj.TestParam,
root.SubObj.SubObj.TestParam)
self.assertEqual(Vals(), (5, 5, 5))
exps.Requested = 'a, TestExp1, b'
self.assertEqual(exps.Active, 'TestExp1')
self.assertEqual(Vals(), (1, 5, 5))
exps.Requested = 'TestExp1, TestExp2'
self.assertEqual(exps.Active, 'TestExp1,TestExp2')
self.assertEqual(Vals(), (1, 1, 5))
# Direct writes to data model should succeed.
# TestExp2 uses the value of root.TestParam at setup time, but
# doesn't look at the value afterward, so changing it doesn't affect
# root.SubObj.TestParam.
root.TestParam = 2
self.assertEqual(Vals(), (2, 1, 5))
# Setting via the ExperimentHandle doesn't change the current
# object value, but does save it for later, when the experiment is
# unapplied.
eh.SetExportParam('TestParam', 3)
self.assertEqual(Vals(), (2, 1, 5))
# Changing the set of experiments reapplies them from scratch after
# unapplying them. That means TestExp2 recalculates its value for
# root.SubObj.TestParam, *after* root.TestParam is restored to the
# value it saved while TestExp1 was active.
exps.Requested = 'TestExp2'
self.assertEqual(Vals(), (3, 3, 5))
# Back to defaults.
exps.Requested = ''
root.TestParam = 5
self.assertEqual(Vals(), (5, 5, 5))
# Order is significant.
exps.Requested = 'TestExp1,TestExp2,TestExpOverlap'
self.assertEqual(Vals(), (2, 1, 'Hello'))
exps.Requested = 'TestExpOverlap,TestExp2,TestExp1'
self.assertEqual(Vals(), (1, 2, 'Hello'))
# You can apply the same experiment more than once.
exps.Requested = 'TestExp1,TestExp2,TestExpOverlap,TestExp1'
self.assertEqual(Vals(), (1, 1, 'Hello'))
def testSysExperiments(self):
# Requested system experiments at startup time are used to pre-populate
# the Requested field.
open(os.path.join(experiment.ACTIVEDIR, 'Splat1.requested'), 'w')
open(os.path.join(experiment.REGDIR, 'Splat2.available'), 'w')
open(os.path.join(experiment.ACTIVEDIR, 'Splat3.active'), 'w')
# ...but if a requested system experiment overlaps with a cwmpd
# experiment, it won't count, because cwmpd experiments don't persist
# across restarts.
open(os.path.join(experiment.ACTIVEDIR, 'TestExp2.requested'), 'w')
root = core.Exporter()
eh = experiment.ExperimentHandle(root)
exps = experiment.Experiments(eh)
eh.root_experiments = exps
self.assertEqual(exps.Requested, 'Splat1,Splat3')
self.assertEqual(exps.Available,
'Splat2,TestExp1,TestExp2,TestExpOverlap')
open(os.path.join(experiment.REGDIR, 'Foo.available'), 'w')
self.assertEqual(exps.Requested, 'Splat1,Splat3')
self.assertEqual(exps.Available,
'Foo,Splat2,TestExp1,TestExp2,TestExpOverlap')
self.assertEqual(exps.Active, '') # Splat3 is active, but not available
open(os.path.join(experiment.ACTIVEDIR, 'Foo.requested'), 'w')
self.assertEqual(exps.Available,
'Foo,Splat2,TestExp1,TestExp2,TestExpOverlap')
self.assertEqual(exps.Active, '')
open(os.path.join(experiment.ACTIVEDIR, 'Foo.active'), 'w')
self.assertEqual(exps.Available,
'Foo,Splat2,TestExp1,TestExp2,TestExpOverlap')
self.assertEqual(exps.Active, 'Foo')
os.unlink(os.path.join(experiment.REGDIR, 'Foo.available'))
self.assertEqual(exps.Available,
'Splat2,TestExp1,TestExp2,TestExpOverlap')
self.assertEqual(exps.Active, '')
os.unlink(os.path.join(experiment.ACTIVEDIR, 'Foo.requested'))
os.unlink(os.path.join(experiment.ACTIVEDIR, 'Foo.active'))
open(os.path.join(experiment.REGDIR, 'Goo.available'), 'w')
self.assertEqual(exps.Available,
'Goo,Splat2,TestExp1,TestExp2,TestExpOverlap')
self.assertEqual(exps.Active, '')
exps.Requested = 'Foo, Goo '
self.assertEqual(exps.Active, '')
self.assertTrue(os.path.exists(
os.path.join(experiment.ACTIVEDIR, 'Foo.requested')))
os.rename(os.path.join(experiment.ACTIVEDIR, 'Goo.requested'),
os.path.join(experiment.ACTIVEDIR, 'Goo.active'))
self.assertEqual(exps.Requested, 'Foo,Goo')
self.assertEqual(exps.Active, 'Goo')
exps.Requested = 'Foo'
self.assertEqual(exps.Active, 'Goo')
os.unlink(os.path.join(experiment.ACTIVEDIR, 'Goo.active'))
self.assertEqual(exps.Requested, 'Foo')
self.assertEqual(exps.Active, '')
os.unlink(os.path.join(experiment.ACTIVEDIR, 'Goo.unrequested'))
self.assertEqual(exps.Requested, 'Foo')
self.assertEqual(exps.Active, '')
def testSysExperimentsInvalidCharacters(self):
root = core.Exporter()
eh = experiment.ExperimentHandle(root)
exps = experiment.Experiments(eh)
eh.root_experiments = exps
exps.Requested = 'Foo/temp.test, Goo_-89'
self.assertFalse(os.path.exists(
os.path.join(experiment.ACTIVEDIR, 'Foo/temp.test.requested')))
self.assertTrue(os.path.exists(
os.path.join(experiment.ACTIVEDIR, 'Goo_-89.requested')))
def testSysExperimentsLongName(self):
root = core.Exporter()
eh = experiment.ExperimentHandle(root)
exps = experiment.Experiments(eh)
eh.root_experiments = exps
long_experiment = 'x' * 128
too_long_experiment = 'y' * 129
exps.Requested = '%s, %s' % (long_experiment, too_long_experiment)
self.assertTrue(os.path.exists(
os.path.join(experiment.ACTIVEDIR, long_experiment + '.requested')))
self.assertFalse(os.path.exists(
os.path.join(experiment.ACTIVEDIR, too_long_experiment + '.requested')))
def testSysExperimentsMaxFiles(self):
root = core.Exporter()
eh = experiment.ExperimentHandle(root)
exps = experiment.Experiments(eh)
eh.root_experiments = exps
for _ in range(256):
temp = tempfile.NamedTemporaryFile(dir=experiment.ACTIVEDIR, delete=False)
temp.close()
exps.Requested = 'Frue'
self.assertFalse(os.path.exists(
os.path.join(experiment.ACTIVEDIR, 'Frue.requested')))
def testSysExperimentsExistAtStartup(self):
open(os.path.join(experiment.ACTIVEDIR, 'Foo.active'), 'w')
open(os.path.join(experiment.ACTIVEDIR, 'Goo.requested'), 'w')
open(os.path.join(experiment.ACTIVEDIR, 'Grue.active'), 'w')
open(os.path.join(experiment.ACTIVEDIR, 'Grue.unrequested'), 'w')
root = core.Exporter()
eh = experiment.ExperimentHandle(root)
exps = experiment.Experiments(eh)
eh.root_experiments = exps
self.assertEqual(exps.Requested, 'Foo,Goo')
self.assertEqual(exps.Available, 'TestExp1,TestExp2,TestExpOverlap')
self.assertEqual(exps.Active, '')
open(os.path.join(experiment.REGDIR, 'Foo.available'), 'w')
self.assertEqual(exps.Active, 'Foo')
open(os.path.join(experiment.REGDIR, 'Goo.available'), 'w')
self.assertEqual(exps.Active, 'Foo')
open(os.path.join(experiment.REGDIR, 'Grue.available'), 'w')
self.assertEqual(exps.Active, 'Foo,Grue')
self.assertEqual(exps.Requested, 'Foo,Goo')
if __name__ == '__main__':
unittest.main()