1# Licensed under the Apache License, Version 2.0 (the "License"); you may
2# not use this file except in compliance with the License. You may obtain
3# a copy of the License at
4#
5#      http://www.apache.org/licenses/LICENSE-2.0
6#
7# Unless required by applicable law or agreed to in writing, software
8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10# License for the specific language governing permissions and limitations
11# under the License.
12
13from unittest import mock
14
15from keystoneauth1 import adapter
16
17from openstack import exceptions
18from openstack.block_storage.v2 import backup
19from openstack.tests.unit import base
20
21FAKE_ID = "6685584b-1eac-4da6-b5c3-555430cf68ff"
22
23BACKUP = {
24    "availability_zone": "az1",
25    "container": "volumebackups",
26    "created_at": "2018-04-02T10:35:27.000000",
27    "updated_at": "2018-04-03T10:35:27.000000",
28    "description": 'description',
29    "fail_reason": 'fail reason',
30    "id": FAKE_ID,
31    "name": "backup001",
32    "object_count": 22,
33    "size": 1,
34    "status": "available",
35    "volume_id": "e5185058-943a-4cb4-96d9-72c184c337d6",
36    "is_incremental": True,
37    "has_dependent_backups": False
38}
39
40
41class TestBackup(base.TestCase):
42
43    def setUp(self):
44        super(TestBackup, self).setUp()
45        self.resp = mock.Mock()
46        self.resp.body = None
47        self.resp.json = mock.Mock(return_value=self.resp.body)
48        self.resp.headers = {}
49        self.resp.status_code = 202
50
51        self.sess = mock.Mock(spec=adapter.Adapter)
52        self.sess.get = mock.Mock()
53        self.sess.post = mock.Mock(return_value=self.resp)
54        self.sess.default_microversion = None
55
56    def test_basic(self):
57        sot = backup.Backup(BACKUP)
58        self.assertEqual("backup", sot.resource_key)
59        self.assertEqual("backups", sot.resources_key)
60        self.assertEqual("/backups", sot.base_path)
61        self.assertTrue(sot.allow_create)
62        self.assertTrue(sot.allow_delete)
63        self.assertTrue(sot.allow_list)
64        self.assertTrue(sot.allow_get)
65        self.assertTrue(sot.allow_fetch)
66
67        self.assertDictEqual(
68            {
69                "all_tenants": "all_tenants",
70                "limit": "limit",
71                "marker": "marker",
72                "name": "name",
73                "project_id": "project_id",
74                "sort_dir": "sort_dir",
75                "sort_key": "sort_key",
76                "status": "status",
77                "volume_id": "volume_id"
78            },
79            sot._query_mapping._mapping
80        )
81
82    def test_create(self):
83        sot = backup.Backup(**BACKUP)
84        self.assertEqual(BACKUP["id"], sot.id)
85        self.assertEqual(BACKUP["name"], sot.name)
86        self.assertEqual(BACKUP["status"], sot.status)
87        self.assertEqual(BACKUP["container"], sot.container)
88        self.assertEqual(BACKUP["availability_zone"], sot.availability_zone)
89        self.assertEqual(BACKUP["created_at"], sot.created_at)
90        self.assertEqual(BACKUP["updated_at"], sot.updated_at)
91        self.assertEqual(BACKUP["description"], sot.description)
92        self.assertEqual(BACKUP["fail_reason"], sot.fail_reason)
93        self.assertEqual(BACKUP["volume_id"], sot.volume_id)
94        self.assertEqual(BACKUP["object_count"], sot.object_count)
95        self.assertEqual(BACKUP["is_incremental"], sot.is_incremental)
96        self.assertEqual(BACKUP["size"], sot.size)
97        self.assertEqual(BACKUP["has_dependent_backups"],
98                         sot.has_dependent_backups)
99
100    def test_create_incremental(self):
101        sot = backup.Backup(is_incremental=True)
102        sot2 = backup.Backup(is_incremental=False)
103
104        create_response = mock.Mock()
105        create_response.status_code = 200
106        create_response.json.return_value = {}
107        create_response.headers = {}
108        self.sess.post.return_value = create_response
109
110        sot.create(self.sess)
111        self.sess.post.assert_called_with(
112            '/backups',
113            headers={},
114            json={
115                'backup': {
116                    'incremental': True,
117                }
118            },
119            microversion=None,
120            params={}
121        )
122
123        sot2.create(self.sess)
124        self.sess.post.assert_called_with(
125            '/backups',
126            headers={},
127            json={
128                'backup': {
129                    'incremental': False,
130                }
131            },
132            microversion=None,
133            params={}
134        )
135
136    def test_restore(self):
137        sot = backup.Backup(**BACKUP)
138
139        self.assertEqual(sot, sot.restore(self.sess, 'vol', 'name'))
140
141        url = 'backups/%s/restore' % FAKE_ID
142        body = {"restore": {"volume_id": "vol", "name": "name"}}
143        self.sess.post.assert_called_with(url, json=body)
144
145    def test_restore_name(self):
146        sot = backup.Backup(**BACKUP)
147
148        self.assertEqual(sot, sot.restore(self.sess, name='name'))
149
150        url = 'backups/%s/restore' % FAKE_ID
151        body = {"restore": {"name": "name"}}
152        self.sess.post.assert_called_with(url, json=body)
153
154    def test_restore_vol_id(self):
155        sot = backup.Backup(**BACKUP)
156
157        self.assertEqual(sot, sot.restore(self.sess, volume_id='vol'))
158
159        url = 'backups/%s/restore' % FAKE_ID
160        body = {"restore": {"volume_id": "vol"}}
161        self.sess.post.assert_called_with(url, json=body)
162
163    def test_restore_no_params(self):
164        sot = backup.Backup(**BACKUP)
165
166        self.assertRaises(
167            exceptions.SDKException,
168            sot.restore,
169            self.sess
170        )
171