source: anuga_validation/automated_validation_tests/patong_validation/validate_patong.py @ 6649

Last change on this file since 6649 was 6649, checked in by rwilson, 16 years ago

Can unpack remote-copied files now, on to setup to run simulation.
File size: 6.7 KB
Line 
1'''
2Automatic verification that the ANUGA code validates against the patong
3dataset as expected.
4
5Required files are downloaded from the ANUGA servers if they are
6out of date or missing.
7'''
8
9import os
10import glob
11import unittest
12#import logging
13import time
14import project
15from anuga.utilities.system_tools import get_web_file, untar_file
16import anuga.utilities.log as log
17
18
19# path to the local data directory
20Local_Data_Directory = os.path.join('.', 'local_data')
21
22# path to the remote data directory
23Remote_Data_Directory = os.path.join('.', 'remote_data')
24
25# sequence of required local data objects
26Local_Data_Objects = ('patong.sww', 'anuga')
27
28# base URL for the remote ANUGA data
29ANUGA_URL = 'http://10.7.64.243/patong_validation_data/'
30
31def update_local_data():
32    # update one data object
33    def update_object(obj, auth):
34        # Get a unique date+time string to defeat caching.  The idea is to add
35        # this to the end of any URL so proxy sees a different request.
36        cache_defeat = '?' + time.strftime('%Y%m%d%H%M%S')
37
38        # create local and remote paths, URLs, etc.
39        remote_path = os.path.join(Remote_Data_Directory, obj)
40        remote_digest = remote_path + '.tgz.digest'
41       
42        local_path = os.path.join(Local_Data_Directory, obj)
43        local_file = local_path + '.tgz'
44        local_digest = local_file + '.digest'
45       
46        object_url = ANUGA_URL + obj + '.tgz'
47        digest_url = object_url + '.digest'
48
49        # see if have both digest and object .tgz
50        if not os.path.exists(local_digest) or not os.path.exists(local_file):
51            # no digest or no object, download both digest and object
52            log.debug('Fetching remote file %s' % digest_url)
53            auth = get_web_file(digest_url+cache_defeat,
54                                local_digest, auth=auth)
55            log.debug('Fetching remote file %s' % object_url)
56            auth = get_web_file(object_url+cache_defeat, local_file, auth=auth)
57        else:
58            # download object digest to remote data directory
59            object_url = ANUGA_URL + obj + '.tgz.digest'
60            log.debug('Fetching remote file %s' % object_url)
61            auth = get_web_file(object_url+cache_defeat,
62                                remote_digest, auth=auth)
63           
64            # compare remote with local digest
65            fd = open(local_digest, 'r')
66            local_digest_data = fd.read()
67            fd.close()
68
69            fd = open(remote_digest, 'r')
70            remote_digest_data = fd.read()
71            fd.close()
72
73            # if local digest != remote digest, refresh object
74            if local_digest_data != remote_digest_data:
75                log.debug('Local file %s is out of date' % obj)
76                fd = open(local_digest, 'w')
77                fd.write(remote_digest_data)
78                fd.close()
79
80                object_url = ANUGA_URL + obj + '.tgz'
81                local_file = local_path + '.tgz'
82                log.debug('Fetching remote file %s' % object_url)
83                auth = get_web_file(object_url+cache_defeat,
84                                    local_file, auth=auth)
85        return auth
86
87   
88    # create local data directory if required
89    if not os.path.exists(Local_Data_Directory):
90        os.mkdir(Local_Data_Directory)
91
92    # create or clean out remote data copy directory
93    if not os.path.exists(Remote_Data_Directory):
94        os.mkdir(Remote_Data_Directory)
95    rem_files = glob.glob(os.path.join(Remote_Data_Directory, '*'))
96    for file in rem_files:
97        os.remove(file)
98
99    # refresh local files
100    auth = None
101    for data_object in Local_Data_Objects:
102        auth = update_object(data_object, auth)
103
104    # unpack *.tgz files
105    for data_object in Local_Data_Objects:
106        tar_path = os.path.join(Local_Data_Directory, data_object+'.tgz')
107        log.debug('Untarring %s in dir %s' % (tar_path, Local_Data_Directory))
108        untar_file(tar_path, target_dir=Local_Data_Directory)
109       
110
111class Test_Patong(unittest.TestCase):
112    def setUp(self):
113        # Check that environment variables are defined.
114        if os.getenv(project.ENV_INUNDATIONHOME) is None:
115            msg = ("Environment variable '%s' is not set in project.py"
116                   % project.ENV_INUNDATIONHOME)
117            raise Exception, msg
118
119        if os.getenv(project.ENV_MUXHOME) is None:
120            msg = ("Environment variable '%s' is not set in project.py"
121                   % project.ENV_MUXHOME)
122            raise Exception, msg
123
124    def tearDown(self):
125        pass
126##        # clear all data objects from local data directory
127##        for data_object in Local_Data_Objects:
128##            try:
129##                log.debug('Deleting %s' % os.path.join(Local_Data_Directory,
130##                                                       data_object))
131##                os.remove(os.path.join(Local_Data_Directory, data_object))
132##            except IOError:
133##                pass
134
135    def test_that_output_is_as_expected(self):
136        # make sure local data is up to date
137        update_local_data()
138        print 'update done'
139
140        # modify environment so we use the local data
141        # INUNDATIONHOME points into the 'anuga' local data directory
142        # N:\georisk_models\inundation\data\thailand\patong_tsunami_scenario\anuga
143        old_inundationhome = os.getenv(project.ENV_INUNDATIONHOME)
144        os.putenv(project.ENV_INUNDATIONHOME,
145                  os.path.join(Local_Data_Directory, 'anuga'))
146        # MUXHOME isn't used, but must exist
147        undo_muxhome = False
148        if os.getenv(project.ENV_MUXHOME) is None:
149            os.putenv(project.ENV_MUXHOME,
150                      os.path.join(Local_Data_Directory, 'data'))
151            undo_muxhome = True
152       
153        # run the simulation, produce SWW file
154        s = 'run_model.py'
155        cmd = 'python %s > %s.stdout' % (s, s)
156        print 'cmd=%s' % cmd
157        res = os.system(cmd)
158        print 'res=%s' % str(res)
159        assert res == 0
160
161        # is SWW file as expected?
162        print 'About to run compare_output_with_expected.py'
163        s = 'compare_output_with_expected.py'
164        res = os.system('python %s > %s.stdout'
165                        % (s, s))
166        print 'Result from %s is %d' % (s, res)
167        assert res == 0
168
169        # undo environment changes
170        if old_inundationhome:
171            os.putenv(project.ENV_INUNDATIONHOME, old_inundationhome)
172        if undo_muxhome:
173            os.putenv(project.ENV_MUXHOME, None)
174
175################################################################################
176
177if __name__ == "__main__":
178    # set logging levels
179    #log.console_logging_level = log.CRITICAL   # no console logging
180    log.console_logging_level = log.DEBUG
181
182    suite = unittest.makeSuite(Test_Patong, 'test')
183    runner = unittest.TextTestRunner(verbosity=2)
184    runner.run(suite)
Note: See TracBrowser for help on using the repository browser.