600-applications-camera.py 10.2 KB
Newer Older
1
#!/usr/bin/env python3
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
#
# Copyright (C) 2011, Nokia Corporation <ivan.frade@nokia.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.
#

"""
Tests trying to simulate the behaviour of applications working with tracker
"""

25
import os
26 27
import random

28
import unittest as ut
29
from common.utils.applicationstest import CommonTrackerApplicationTest as CommonTrackerApplicationTest
30
from common.utils.helpers import log
31 32


33 34 35 36
NMM_PHOTO = 'http://www.tracker-project.org/temp/nmm#Photo'
NMM_VIDEO = 'http://www.tracker-project.org/temp/nmm#Video'


37 38 39 40 41
class TrackerCameraTestSuite (CommonTrackerApplicationTest):
    """
    Common functionality for camera tests.
    """

42
    def insert_photo_resource_info(self, urn, file_url):
43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72
        """
        Insert new photo resource in the store, including nie:mimeType and nie:url
        """
        insert = """
        INSERT { <%(urn)s>
            a nie:InformationElement,
            nie:DataObject,
            nfo:Image,
            nfo:Media,
            nfo:Visual,
            nmm:Photo
        }

        DELETE { <%(urn)s> nie:mimeType ?_1 }
        WHERE { <%(urn)s> nie:mimeType ?_1 }

        INSERT { <%(urn)s>
            a rdfs:Resource ;
            nie:mimeType \"image/jpeg\"
        }

        DELETE { <%(urn)s> nie:url ?_2 }
        WHERE { <%(urn)s> nie:url ?_2 }

        INSERT { <%(urn)s>
            a rdfs:Resource ;
            nie:url \"%(file_url)s\" ;
            nie:isStoredAs <%(urn)s>
        }
        """ % locals()
73 74
        self.tracker.update(insert)
        self.assertEqual(self.get_urn_count_by_url(file_url), 1)
75

76
    def insert_video_resource_info(self, urn, file_url):
77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106
        """
        Insert new video resource in the store, including nie:mimeType and nie:url
        """
        insert = """
        INSERT { <%(urn)s>
            a nie:InformationElement,
            nie:DataObject,
            nfo:Video,
            nfo:Media,
            nfo:Visual,
            nmm:Video
        }

        DELETE { <%(urn)s> nie:mimeType ?_1 }
        WHERE { <%(urn)s> nie:mimeType ?_1 }

        INSERT { <%(urn)s>
            a rdfs:Resource ;
            nie:mimeType \"video/mp4\"
        }

        DELETE { <%(urn)s> nie:url ?_2 }
        WHERE { <%(urn)s> nie:url ?_2 }

        INSERT { <%(urn)s>
            a rdfs:Resource ;
            nie:url \"%(file_url)s\" ;
            nie:isStoredAs <%(urn)s>
        }
        """ % locals()
107 108
        self.tracker.update(insert)
        self.assertEqual(self.get_urn_count_by_url(file_url), 1)
109

110
    def insert_dummy_location_info(self, fileurn, geolocationurn, postaladdressurn):
111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127
        """
        Insert placeholder location info for a file
        """
        location_insert = """
        INSERT { <%s> a             nco:PostalAddress ;
                      nco:country  \"SPAIN\" ;
                      nco:locality \"Tres Cantos\"
        }

        INSERT { <%s> a                 slo:GeoLocation ;
                      slo:postalAddress <%s>
        }

        INSERT { <%s> a            rdfs:Resource ;
                      slo:location <%s>
        }
        """ % (postaladdressurn, geolocationurn, postaladdressurn, fileurn, geolocationurn)
128
        self.tracker.update(location_insert)
129 130 131


class TrackerCameraPicturesApplicationTests (TrackerCameraTestSuite):
132

133
    def test_01_camera_picture(self):
134 135 136 137 138 139 140 141 142
        """
        Camera simulation:

        1. Create resource in the store for the new file
        2. Write the file
        3. Wait for miner-fs to index it
        4. Ensure no duplicates are found
        """

143 144 145
        fileurn = "tracker://test_camera_picture_01/" + str(random.randint(0, 100))
        origin_filepath = os.path.join(self.get_data_dir(), self.get_test_image())
        dest_filepath = os.path.join(self.get_dest_dir(), self.get_test_image())
146
        dest_fileuri = "file://" + dest_filepath
147

148
        self.insert_photo_resource_info(fileurn, dest_fileuri)
149

150
        # Copy the image to the dest path
151 152 153 154
        self.slowcopy_file(origin_filepath, dest_filepath)
        assert os.path.exists(dest_filepath)
        dest_id, dest_urn = self.system.store.await_resource_inserted(NMM_PHOTO, dest_fileuri)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 1)
155 156

        # Clean the new file so the test directory is as before
157 158 159 160
        log("Remove and wait")
        os.remove(dest_filepath)
        self.system.store.await_resource_deleted(NMM_PHOTO, dest_id)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 0)
161

162
    def test_02_camera_picture_geolocation(self):
163 164 165 166 167 168 169 170 171
        """
        Camera simulation:

        1. Create resource in the store for the new file
        2. Set nlo:location
        2. Write the file
        3. Wait for miner-fs to index it
        4. Ensure no duplicates are found
        """
172

173 174
        fileurn = "tracker://test_camera_picture_02/" + str(random.randint(0, 100))
        dest_filepath = os.path.join(self.get_dest_dir(), self.get_test_image())
175
        dest_fileuri = "file://" + dest_filepath
176

177 178
        geolocationurn = "tracker://test_camera_picture_02_geolocation/" + str(random.randint(0, 100))
        postaladdressurn = "tracker://test_camera_picture_02_postaladdress/" + str(random.randint(0, 100))
179

180
        self.insert_photo_resource_info(fileurn, dest_fileuri)
181

182 183
        # FIRST, open the file for writing, and just write some garbage, to simulate that
        # we already started recording the video...
184 185 186 187
        fdest = open(dest_filepath, 'wb')
        fdest.write("some garbage written here")
        fdest.write("to simulate we're recording something...")
        fdest.seek(0)
188

189
        # SECOND, set slo:location
190
        self.insert_dummy_location_info(fileurn, geolocationurn, postaladdressurn)
191

192 193 194 195 196
        # THIRD, start copying the image to the dest path
        original_file = os.path.join(self.get_data_dir(), self.get_test_image())
        self.slowcopy_file_fd(original_file, fdest)
        fdest.close()
        assert os.path.exists(dest_filepath)
197 198

        # FOURTH, ensure we have only 1 resource
199 200
        dest_id, dest_urn = self.system.store.await_resource_inserted(NMM_PHOTO, dest_fileuri)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 1)
201 202

        # Clean the new file so the test directory is as before
203 204 205 206
        log("Remove and wait")
        os.remove(dest_filepath)
        self.system.store.await_resource_deleted(NMM_PHOTO, dest_id)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 0)
207 208


209
class TrackerCameraVideosApplicationTests (TrackerCameraTestSuite):
210

211
    def test_01_camera_video(self):
212 213 214 215 216 217 218 219 220
        """
        Camera video recording simulation:

        1. Create resource in the store for the new file
        2. Write the file
        3. Wait for miner-fs to index it
        4. Ensure no duplicates are found
        """

221 222 223
        fileurn = "tracker://test_camera_video_01/" + str(random.randint(0, 100))
        origin_filepath = os.path.join(self.get_data_dir(), self.get_test_video())
        dest_filepath = os.path.join(self.get_dest_dir(), self.get_test_video())
224
        dest_fileuri = "file://" + dest_filepath
225

226
        self.insert_video_resource_info(fileurn, dest_fileuri)
227

228
        # Copy the image to the dest path
229 230 231 232
        self.slowcopy_file(origin_filepath, dest_filepath)
        assert os.path.exists(dest_filepath)
        dest_id, dest_urn = self.system.store.await_resource_inserted(NMM_PHOTO, dest_fileuri)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 1)
233 234

        # Clean the new file so the test directory is as before
235 236 237 238
        log("Remove and wait")
        os.remove(dest_filepath)
        self.system.store.await_resource_deleted(NMM_PHOTO, dest_id)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 0)
239

240
    def test_02_camera_video_geolocation(self):
241 242 243 244 245 246 247 248 249 250
        """
        Camera simulation:

        1. Create resource in the store for the new file
        2. Set nlo:location
        2. Write the file
        3. Wait for miner-fs to index it
        4. Ensure no duplicates are found
        """

251 252 253
        fileurn = "tracker://test_camera_video_02/" + str(random.randint(0, 100))
        origin_filepath = os.path.join(self.get_data_dir(), self.get_test_video())
        dest_filepath = os.path.join(self.get_dest_dir(), self.get_test_video())
254
        dest_fileuri = "file://" + dest_filepath
255

256 257
        geolocationurn = "tracker://test_camera_video_02_geolocation/" + str(random.randint(0, 100))
        postaladdressurn = "tracker://test_camera_video_02_postaladdress/" + str(random.randint(0, 100))
258

259
        self.insert_video_resource_info(fileurn, dest_fileuri)
260

261 262
        # FIRST, open the file for writing, and just write some garbage, to simulate that
        # we already started recording the video...
263 264 265 266
        fdest = open(dest_filepath, 'wb')
        fdest.write("some garbage written here")
        fdest.write("to simulate we're recording something...")
        fdest.seek(0)
267

268
        # SECOND, set slo:location
269
        self.insert_dummy_location_info(fileurn, geolocationurn, postaladdressurn)
270

271 272 273 274
        # THIRD, start copying the image to the dest path
        self.slowcopy_file_fd(origin_filepath, fdest)
        fdest.close()
        assert os.path.exists(dest_filepath)
275 276

        # FOURTH, ensure we have only 1 resource
277 278
        dest_id, dest_urn = self.system.store.await_resource_inserted(NMM_VIDEO, dest_fileuri)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 1)
279

280
        # Clean the new file so the test directory is as before
281 282 283 284
        log("Remove and wait")
        os.remove(dest_filepath)
        self.system.store.await_resource_deleted(NMM_VIDEO, dest_id)
        self.assertEqual(self.get_urn_count_by_url(dest_fileuri), 0)
285 286


287 288
if __name__ == "__main__":
    ut.main()