← Back to branch summary

~thomir-deactivatedaccount/adt-continuous-deployer/trunk-fix-warning-message

« back to all changes in this revision

Viewing changes to ci_automation/tests/test_mojo_project_pool.py

  • Committer: Ubuntu CI Bot
  • Author(s): Thomi Richards
  • Date: 2015-05-28 06:22:24 UTC
  • mfrom: (54.2.3 trunk-add-mojo-project-pool)
  • Revision ID: ubuntu_ci_bot-20150528062224-ke6exobip6l9iu8f
Add mojo project pool code to cd.py. [r=Thomi Richards, Francis Ginther]

Show diffs side-by-side

added added

removed removed

Lines of Context:
ci_automation/tests/test_mojo_project_pool.py
 
1
#!/usr/bin/env python3
 
2
#
 
3
# Copyright (C) 2015 Canonical
 
4
#
 
5
# This program is free software: you can redistribute it and/or modify
 
6
# it under the terms of the GNU General Public License as published by
 
7
# the Free Software Foundation, either version 3 of the License, or
 
8
# (at your option) any later version.
 
9
#
 
10
# This program is distributed in the hope that it will be useful,
 
11
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
12
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
13
# GNU General Public License for more details.
 
14
#
 
15
# You should have received a copy of the GNU General Public License
 
16
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
17
#
 
18
 
 
19
import os.path
 
20
import tempfile
 
21
import unittest
 
22
import shutil
 
23
 
 
24
from ci_automation.mojo_project_pool import (
 
25
    MojoProjectPool,
 
26
    NoProjectError,
 
27
)
 
28
 
 
29
 
 
30
class MojoProjectPoolTests(unittest.TestCase):
 
31
 
 
32
    def test_creates_project_and_lock_directories(self):
 
33
        with tempfile.TemporaryDirectory() as tempdir:
 
34
            project_path = os.path.join(tempdir, 'projects')
 
35
            lock_path = os.path.join(tempdir, 'projects', 'locks')
 
36
            pool = MojoProjectPool(project_path, lock_path)
 
37
 
 
38
            self.assertTrue(os.path.exists(project_path))
 
39
            self.assertTrue(os.path.exists(lock_path))
 
40
 
 
41
    def test_cannot_acquire_project_when_none_are_setup(self):
 
42
        with ProjectPoolDir() as workdir:
 
43
            pool = MojoProjectPool(workdir.project_dir, workdir.lock_dir)
 
44
 
 
45
            with self.assertRaises(NoProjectError):
 
46
                with pool.acquire_mojo_project():
 
47
                    pass
 
48
 
 
49
    def test_can_acquire_single_project(self):
 
50
        with ProjectPoolDir(projects=['project-one']) as workdir:
 
51
            pool = MojoProjectPool(workdir.project_dir,workdir.lock_dir)
 
52
            with pool.acquire_mojo_project() as project:
 
53
                self.assertEqual(project, 'project-one')
 
54
 
 
55
    def test_can_acquire_single_project_once_only(self):
 
56
        with ProjectPoolDir(projects=['project-one']) as workdir:
 
57
            pool = MojoProjectPool(workdir.project_dir,workdir.lock_dir)
 
58
            with pool.acquire_mojo_project() as project:
 
59
                self.assertEqual(project, 'project-one')
 
60
                with self.assertRaises(NoProjectError):
 
61
                    with pool.acquire_mojo_project():
 
62
                        pass
 
63
 
 
64
    def test_can_acquire_multiple_projects(self):
 
65
        with ProjectPoolDir(projects=['one', 'two']) as workdir:
 
66
            pool = MojoProjectPool(workdir.project_dir,workdir.lock_dir)
 
67
            with pool.acquire_mojo_project() as p1, pool.acquire_mojo_project() as p2:
 
68
                self.assertNotEqual(p1, p2)
 
69
                self.assertIn(p1, ('one', 'two'))
 
70
                self.assertIn(p2, ('one', 'two'))
 
71
 
 
72
    def test_uncaught_exception_releases_lock(self):
 
73
        with ProjectPoolDir(projects=['project-one']) as workdir:
 
74
            pool = MojoProjectPool(workdir.project_dir,workdir.lock_dir)
 
75
            try:
 
76
                with pool.acquire_mojo_project() as project:
 
77
                    raise Exception("Pretending mojo failed!")
 
78
            except Exception:
 
79
                # must be able to re-acquire the project (it must have been
 
80
                # released).
 
81
                with pool.acquire_mojo_project() as project:
 
82
                    self.assertEqual(project, 'project-one')
 
83
 
 
84
 
 
85
class ProjectPoolDir(object):
 
86
 
 
87
    """A context manager that creates a project pool on disk."""
 
88
    def __init__(self, projects=[]):
 
89
        self.project_dir = None
 
90
        self.lock_dir = None
 
91
        self._create_projects = projects
 
92
 
 
93
    def __enter__(self):
 
94
        self.workdir = tempfile.mkdtemp()
 
95
        self.project_dir = os.path.join(self.workdir, 'projects')
 
96
        self.lock_dir = os.path.join(self.workdir, 'locks')
 
97
        if self._create_projects:
 
98
            os.makedirs(self.project_dir)
 
99
            for project_name in self._create_projects:
 
100
                with open(os.path.join(self.project_dir, project_name), 'w'):
 
101
                    pass
 
102
        return self
 
103
 
 
104
    def __exit__(self, exc_type, exc_value, traceback):
 
105
        shutil.rmtree(self.workdir)
 
106
        self.project_dir = None
 
107
        self.lock_dir = None
 
108
 
 
109
 
 
110
if __name__ == '__main__':
 
111
    unittest.main()