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.v3 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    "os-backup-project-attr:project_id": "2c67a14be9314c5dae2ee6c4ec90cf0b",
39    "user_id": "515ba0dd59f84f25a6a084a45d8d93b2",
40    "metadata": {"key": "value"}
41}
42
43
44class TestBackup(base.TestCase):
45
46    def setUp(self):
47        super(TestBackup, self).setUp()
48        self.resp = mock.Mock()
49        self.resp.body = None
50        self.resp.json = mock.Mock(return_value=self.resp.body)
51        self.resp.headers = {}
52        self.resp.status_code = 202
53
54        self.sess = mock.Mock(spec=adapter.Adapter)
55        self.sess.get = mock.Mock()
56        self.sess.post = mock.Mock(return_value=self.resp)
57        self.sess.default_microversion = None
58
59    def test_basic(self):
60        sot = backup.Backup(BACKUP)
61        self.assertEqual("backup", sot.resource_key)
62        self.assertEqual("backups", sot.resources_key)
63        self.assertEqual("/backups", sot.base_path)
64        self.assertTrue(sot.allow_create)
65        self.assertTrue(sot.allow_delete)
66        self.assertTrue(sot.allow_list)
67        self.assertTrue(sot.allow_get)
68        self.assertTrue(sot.allow_fetch)
69
70        self.assertDictEqual(
71            {
72                "all_tenants": "all_tenants",
73                "limit": "limit",
74                "marker": "marker",
75                "name": "name",
76                "project_id": "project_id",
77                "sort_dir": "sort_dir",
78                "sort_key": "sort_key",
79                "status": "status",
80                "volume_id": "volume_id"
81            },
82            sot._query_mapping._mapping
83        )
84
85    def test_create(self):
86        sot = backup.Backup(**BACKUP)
87        self.assertEqual(BACKUP["id"], sot.id)
88        self.assertEqual(BACKUP["name"], sot.name)
89        self.assertEqual(BACKUP["status"], sot.status)
90        self.assertEqual(BACKUP["container"], sot.container)
91        self.assertEqual(BACKUP["availability_zone"], sot.availability_zone)
92        self.assertEqual(BACKUP["created_at"], sot.created_at)
93        self.assertEqual(BACKUP["updated_at"], sot.updated_at)
94        self.assertEqual(BACKUP["description"], sot.description)
95        self.assertEqual(BACKUP["fail_reason"], sot.fail_reason)
96        self.assertEqual(BACKUP["volume_id"], sot.volume_id)
97        self.assertEqual(BACKUP["object_count"], sot.object_count)
98        self.assertEqual(BACKUP["is_incremental"], sot.is_incremental)
99        self.assertEqual(BACKUP["size"], sot.size)
100        self.assertEqual(BACKUP["has_dependent_backups"],
101                         sot.has_dependent_backups)
102        self.assertEqual(BACKUP['os-backup-project-attr:project_id'],
103                         sot.project_id)
104        self.assertEqual(BACKUP['metadata'], sot.metadata)
105        self.assertEqual(BACKUP['user_id'], sot.user_id)
106
107    def test_create_incremental(self):
108        sot = backup.Backup(is_incremental=True)
109        sot2 = backup.Backup(is_incremental=False)
110
111        create_response = mock.Mock()
112        create_response.status_code = 200
113        create_response.json.return_value = {}
114        create_response.headers = {}
115        self.sess.post.return_value = create_response
116
117        sot.create(self.sess)
118        self.sess.post.assert_called_with(
119            '/backups',
120            headers={},
121            json={
122                'backup': {
123                    'incremental': True,
124                }
125            },
126            microversion=None,
127            params={}
128        )
129
130        sot2.create(self.sess)
131        self.sess.post.assert_called_with(
132            '/backups',
133            headers={},
134            json={
135                'backup': {
136                    'incremental': False,
137                }
138            },
139            microversion=None,
140            params={}
141        )
142
143    def test_restore(self):
144        sot = backup.Backup(**BACKUP)
145
146        self.assertEqual(sot, sot.restore(self.sess, 'vol', 'name'))
147
148        url = 'backups/%s/restore' % FAKE_ID
149        body = {"restore": {"volume_id": "vol", "name": "name"}}
150        self.sess.post.assert_called_with(url, json=body)
151
152    def test_restore_name(self):
153        sot = backup.Backup(**BACKUP)
154
155        self.assertEqual(sot, sot.restore(self.sess, name='name'))
156
157        url = 'backups/%s/restore' % FAKE_ID
158        body = {"restore": {"name": "name"}}
159        self.sess.post.assert_called_with(url, json=body)
160
161    def test_restore_vol_id(self):
162        sot = backup.Backup(**BACKUP)
163
164        self.assertEqual(sot, sot.restore(self.sess, volume_id='vol'))
165
166        url = 'backups/%s/restore' % FAKE_ID
167        body = {"restore": {"volume_id": "vol"}}
168        self.sess.post.assert_called_with(url, json=body)
169
170    def test_restore_no_params(self):
171        sot = backup.Backup(**BACKUP)
172
173        self.assertRaises(
174            exceptions.SDKException,
175            sot.restore,
176            self.sess
177        )
178