| #!/usr/bin/env python |
| # |
| # Copyright 2007 Google Inc. |
| # |
| # 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. |
| # |
| """Tests for google.apphosting.tools.devappserver2.devappserver2.""" |
| |
| |
| |
| import argparse |
| import getpass |
| import itertools |
| import os |
| import os.path |
| import sys |
| import tempfile |
| import unittest |
| |
| import google |
| import mox |
| |
| from google.appengine.tools.devappserver2 import devappserver2 |
| |
| |
| class WinError(Exception): |
| pass |
| |
| |
| class GenerateStoragePathsTest(unittest.TestCase): |
| """Tests for devappserver._generate_storage_paths.""" |
| |
| def setUp(self): |
| self.mox = mox.Mox() |
| self.mox.StubOutWithMock(getpass, 'getuser') |
| self.mox.StubOutWithMock(tempfile, 'gettempdir') |
| |
| def tearDown(self): |
| self.mox.UnsetStubs() |
| |
| @unittest.skipUnless(sys.platform.startswith('win'), 'Windows only') |
| def test_windows(self): |
| tempfile.gettempdir().AndReturn('/tmp') |
| |
| self.mox.ReplayAll() |
| self.assertEqual( |
| [os.path.join('/tmp', 'appengine.myapp'), |
| os.path.join('/tmp', 'appengine.myapp.1'), |
| os.path.join('/tmp', 'appengine.myapp.2')], |
| list(itertools.islice(devappserver2._generate_storage_paths('myapp'), |
| 3))) |
| self.mox.VerifyAll() |
| |
| @unittest.skipIf(sys.platform.startswith('win'), 'not on Windows') |
| def test_working_getuser(self): |
| getpass.getuser().AndReturn('johndoe') |
| tempfile.gettempdir().AndReturn('/tmp') |
| |
| self.mox.ReplayAll() |
| self.assertEqual( |
| [os.path.join('/tmp', 'appengine.myapp.johndoe'), |
| os.path.join('/tmp', 'appengine.myapp.johndoe.1'), |
| os.path.join('/tmp', 'appengine.myapp.johndoe.2')], |
| list(itertools.islice(devappserver2._generate_storage_paths('myapp'), |
| 3))) |
| self.mox.VerifyAll() |
| |
| @unittest.skipIf(sys.platform.startswith('win'), 'not on Windows') |
| def test_broken_getuser(self): |
| getpass.getuser().AndRaise(Exception()) |
| tempfile.gettempdir().AndReturn('/tmp') |
| |
| self.mox.ReplayAll() |
| self.assertEqual( |
| [os.path.join('/tmp', 'appengine.myapp'), |
| os.path.join('/tmp', 'appengine.myapp.1'), |
| os.path.join('/tmp', 'appengine.myapp.2')], |
| list(itertools.islice(devappserver2._generate_storage_paths('myapp'), |
| 3))) |
| self.mox.VerifyAll() |
| |
| |
| class GetStoragePathTest(unittest.TestCase): |
| """Tests for devappserver._get_storage_path.""" |
| |
| def setUp(self): |
| self.mox = mox.Mox() |
| self.mox.StubOutWithMock(devappserver2, '_generate_storage_paths') |
| |
| def tearDown(self): |
| self.mox.UnsetStubs() |
| |
| def test_no_path_given_directory_does_not_exist(self): |
| path = tempfile.mkdtemp() |
| os.rmdir(path) |
| devappserver2._generate_storage_paths('example.com_myapp').AndReturn([path]) |
| |
| self.mox.ReplayAll() |
| self.assertEqual( |
| path, |
| devappserver2._get_storage_path(None, 'dev~example.com:myapp')) |
| self.mox.VerifyAll() |
| self.assertTrue(os.path.isdir(path)) |
| |
| def test_no_path_given_directory_exists(self): |
| path1 = tempfile.mkdtemp() |
| os.chmod(path1, 0777) |
| path2 = tempfile.mkdtemp() # Made with mode 0700. |
| |
| devappserver2._generate_storage_paths('example.com_myapp').AndReturn( |
| [path1, path2]) |
| |
| self.mox.ReplayAll() |
| if sys.platform == 'win32': |
| expected_path = path1 |
| else: |
| expected_path = path2 |
| self.assertEqual( |
| expected_path, |
| devappserver2._get_storage_path(None, 'dev~example.com:myapp')) |
| self.mox.VerifyAll() |
| |
| def test_path_given_does_not_exist(self): |
| path = tempfile.mkdtemp() |
| os.rmdir(path) |
| |
| self.assertEqual( |
| path, |
| devappserver2._get_storage_path(path, 'dev~example.com:myapp')) |
| self.assertTrue(os.path.isdir(path)) |
| |
| def test_path_given_not_directory(self): |
| _, path = tempfile.mkstemp() |
| |
| self.assertRaises( |
| IOError, |
| devappserver2._get_storage_path, path, 'dev~example.com:myapp') |
| |
| def test_path_given_exists(self): |
| path = tempfile.mkdtemp() |
| |
| self.assertEqual( |
| path, |
| devappserver2._get_storage_path(path, 'dev~example.com:myapp')) |
| |
| |
| class PortParserTest(unittest.TestCase): |
| |
| def test_valid_port(self): |
| self.assertEqual(8080, devappserver2.PortParser()('8080')) |
| |
| def test_port_zero_allowed(self): |
| self.assertEqual(0, devappserver2.PortParser()('0')) |
| |
| def test_port_zero_not_allowed(self): |
| self.assertRaises(argparse.ArgumentTypeError, |
| devappserver2.PortParser(allow_port_zero=False), '0') |
| |
| def test_negative_port(self): |
| self.assertRaises(argparse.ArgumentTypeError, devappserver2.PortParser(), |
| '-1') |
| |
| def test_port_too_high(self): |
| self.assertRaises(argparse.ArgumentTypeError, devappserver2.PortParser(), |
| '65536') |
| |
| def test_port_max_value(self): |
| self.assertEqual(65535, devappserver2.PortParser()('65535')) |
| |
| def test_not_an_int(self): |
| self.assertRaises(argparse.ArgumentTypeError, devappserver2.PortParser(), |
| 'a port') |
| |
| |
| class ParseMaxServerInstancesTest(unittest.TestCase): |
| |
| def test_single_valid_arg(self): |
| self.assertEqual(1, devappserver2.parse_max_module_instances('1')) |
| |
| def test_single_zero_arg(self): |
| self.assertRaisesRegexp(argparse.ArgumentTypeError, |
| 'count must be greater than zero', |
| devappserver2.parse_max_module_instances, '0') |
| |
| def test_single_negative_arg(self): |
| self.assertRaisesRegexp(argparse.ArgumentTypeError, |
| 'count must be greater than zero', |
| devappserver2.parse_max_module_instances, '-1') |
| |
| def test_single_nonint_arg(self): |
| self.assertRaisesRegexp(argparse.ArgumentTypeError, |
| 'Invalid max instance count:', |
| devappserver2.parse_max_module_instances, 'cat') |
| |
| def test_multiple_valid_args(self): |
| self.assertEqual( |
| {'default': 10, |
| 'foo': 5}, |
| devappserver2.parse_max_module_instances('default:10,foo:5')) |
| |
| def test_multiple_non_colon(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, |
| 'Expected "module:max_instance_count"', |
| devappserver2.parse_max_module_instances, 'default:10,foo') |
| |
| def test_multiple_non_int(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, |
| 'Expected "module:max_instance_count"', |
| devappserver2.parse_max_module_instances, 'default:cat') |
| |
| def test_duplicate_modules(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, |
| 'Duplicate max instance count', |
| devappserver2.parse_max_module_instances, 'default:5,default:10') |
| |
| def test_multiple_with_zero(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, |
| 'count for module zero must be greater than zero', |
| devappserver2.parse_max_module_instances, 'default:5,zero:0') |
| |
| def test_multiple_with_negative(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, |
| 'count for module negative must be greater than zero', |
| devappserver2.parse_max_module_instances, 'default:5,negative:-1') |
| |
| def test_multiple_missing_name(self): |
| self.assertEqual( |
| {'default': 10}, |
| devappserver2.parse_max_module_instances(':10')) |
| |
| def test_multiple_missing_count(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, |
| 'Expected "module:max_instance_count"', |
| devappserver2.parse_max_module_instances, 'default:') |
| |
| |
| class ParseThreadsafeOverrideTest(unittest.TestCase): |
| |
| def test_single_valid_arg(self): |
| self.assertTrue(devappserver2.parse_threadsafe_override('True')) |
| self.assertFalse(devappserver2.parse_threadsafe_override('No')) |
| |
| def test_single_nonbool_art(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, 'Invalid threadsafe override', |
| devappserver2.parse_threadsafe_override, 'okaydokey') |
| |
| def test_multiple_valid_args(self): |
| self.assertEqual( |
| {'default': False, |
| 'foo': True}, |
| devappserver2.parse_threadsafe_override('default:False,foo:True')) |
| |
| def test_multiple_non_colon(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, 'Expected "module:threadsafe_override"', |
| devappserver2.parse_threadsafe_override, 'default:False,foo') |
| |
| def test_multiple_non_int(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, 'Expected "module:threadsafe_override"', |
| devappserver2.parse_threadsafe_override, 'default:okaydokey') |
| |
| def test_duplicate_modules(self): |
| self.assertRaisesRegexp( |
| argparse.ArgumentTypeError, |
| 'Duplicate threadsafe override value', |
| devappserver2.parse_threadsafe_override, 'default:False,default:True') |
| |
| def test_multiple_missing_name(self): |
| self.assertEqual( |
| {'default': False}, |
| devappserver2.parse_threadsafe_override(':No')) |
| |
| |
| class FakeApplicationConfiguration(object): |
| |
| def __init__(self, modules): |
| self.modules = modules |
| |
| |
| class FakeModuleConfiguration(object): |
| |
| def __init__(self, module_name): |
| self.module_name = module_name |
| |
| |
| class CreateModuleToSettingTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.application_configuration = FakeApplicationConfiguration([ |
| FakeModuleConfiguration('m1'), FakeModuleConfiguration('m2'), |
| FakeModuleConfiguration('m3')]) |
| |
| def testNone(self): |
| self.assertEquals( |
| {}, |
| devappserver2.DevelopmentServer._create_module_to_setting( |
| None, self.application_configuration, '--option')) |
| |
| def testDict(self): |
| self.assertEquals( |
| {'m1': 3, 'm3': 1}, |
| devappserver2.DevelopmentServer._create_module_to_setting( |
| {'m1': 3, 'm3': 1}, self.application_configuration, '--option')) |
| |
| def testSingleValue(self): |
| self.assertEquals( |
| {'m1': True, 'm2': True, 'm3': True}, |
| devappserver2.DevelopmentServer._create_module_to_setting( |
| True, self.application_configuration, '--option')) |
| |
| def testDictWithUnknownModules(self): |
| self.assertEquals( |
| {'m1': 3.5}, |
| devappserver2.DevelopmentServer._create_module_to_setting( |
| {'m1': 3.5, 'm4': 2.7}, self.application_configuration, '--option')) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
