863878245

Build Type

github

Committed by GitHub

Commit Message

Merge 8d34c4c5c into b917f0a72

Pull Request Pull Request #28: Add level1c runner

Run Details

80 of 438 branches covered (18.26%)

Branch coverage included in aggregate %.

157 of 257 new or added lines in 8 files covered. (61.09%)

16 existing lines in 1 file now uncovered.

710 of 1568 relevant lines covered (45.28%)

1.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.31

/nwcsafpps_runner/tests/test_level1c_runner.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Copyright (c) 2021 Adam.Dybbroe

# Author(s):

#   Adam.Dybbroe <a000680@c21856.ad.smhi.se>

# 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 3 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, see <http://www.gnu.org/licenses/>.

"""Unit testing the level-1c runner code
"""

import pytest
from unittest.mock import patch
import unittest
from posttroll.message import Message
from datetime import datetime

from nwcsafpps_runner.message_utils import publish_l1c, prepare_l1c_message
from nwcsafpps_runner.l1c_processing import check_message_okay
from nwcsafpps_runner.l1c_processing import check_service_name_okay
# from nwcsafpps_runner.l1c_processing import L1cProcessor
from nwcsafpps_runner.l1c_processing import ServiceNameNotSupported


class MyFakePublisher(object):

    def __init__(self):
        pass

    def send(self, message):
        pass


TEST_INPUT_MSG = """pytroll://1b/hrit/0deg dataset safusr.u@lxserv1043.smhi.se 2021-05-18T14:28:54.154172 v1.01 application/json {"data_type": "MSG4", "orig_platform_name": "MSG4", "start_time": "2021-05-18T14:15:00", "variant": "0DEG", "series": "MSG4", "platform_name": "Meteosat-11", "channel": "", "nominal_time": "2021-05-18T14:15:00", "compressed": "", "origin": "172.18.0.248:9093", "dataset": [{"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__", "uid": "H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__"}, {"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__", "uid": "H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__"}], "sensor": ["seviri"]}"""

TEST_INPUT_MSG_NO_DATASET = """pytroll://1b/hrit/0deg file safusr.u@lxserv1043.smhi.se 2021-05-18T14:28:54.154172 v1.01 application/json {"data_type": "MSG4", "orig_platform_name": "MSG4", "start_time": "2021-05-18T14:15:00", "variant": "0DEG", "series": "MSG4", "platform_name": "Meteosat-11", "channel": "", "nominal_time": "2021-05-18T14:15:00", "compressed": "", "origin": "172.18.0.248:9093", "file": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__", "sensor": ["seviri"]}"""


TEST_INPUT_MSG_NO_PLATFORM_NAME = """pytroll://1b/hrit/0deg dataset safusr.u@lxserv1043.smhi.se 2021-05-18T14:28:54.154172 v1.01 application/json {"data_type": "MSG4", "orig_platform_name": "MSG4", "start_time": "2021-05-18T14:15:00", "variant": "0DEG", "series": "MSG4", "channel": "", "nominal_time": "2021-05-18T14:15:00", "compressed": "", "origin": "172.18.0.248:9093", "dataset": [{"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__", "uid": "H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__"}, {"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__", "uid": "H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__"}], "sensor": ["seviri"]}"""


class TestPublishMessage(unittest.TestCase):

    @patch('nwcsafpps_runner.message_utils.socket.gethostname')
    def test_create_publish_message(self, gethostname):
        """Test the creation of the publish message."""

        gethostname.return_value = "my_local_server"
        my_fake_level1c_file = '/my/level1c/file/path/level1c.nc'
        input_msg = Message.decode(rawstr=TEST_INPUT_MSG)

        result = prepare_l1c_message(my_fake_level1c_file, input_msg.data, orbit=99999)

        expected = {'data_type': 'MSG4', 'orig_platform_name': 'MSG4',
                    'start_time': datetime(2021, 5, 18, 14, 15),
                    'variant': '0DEG', 'series': 'MSG4',
                    'platform_name': 'Meteosat-11',
                    'channel': '',
                    'nominal_time': datetime(2021, 5, 18, 14, 15),
                    'compressed': '',
                    'origin': '172.18.0.248:9093', 'sensor': ['seviri'],
                    'uri': 'ssh://my_local_server/my/level1c/file/path/level1c.nc',
                    'uid': 'level1c.nc',
                    'format': 'PPS-L1C',
                    'type': 'NETCDF',
                    'data_processing_level': '1c'}

        self.assertDictEqual(result, expected)

    @patch('nwcsafpps_runner.message_utils.Message.encode')
    def test_publish_messages(self, mock_message):
        """Test the sending the messages."""

        my_fake_publisher = MyFakePublisher()
        mock_message.return_value = "some pytroll message"

        pub_message = {'data_type': 'MSG4', 'orig_platform_name': 'MSG4',
                       'start_time': datetime(2021, 5, 18, 14, 15),
                       'variant': '0DEG', 'series': 'MSG4',
                       'platform_name': 'Meteosat-11',
                       'channel': '',
                       'nominal_time': datetime(2021, 5, 18, 14, 15),
                       'compressed': '',
                       'origin': '172.18.0.248:9093', 'sensor': ['seviri'],
                       'uri': 'ssh://my_local_server/my/level1c/file/path/level1c.nc',
                       'uid': 'level1c.nc',
                       'format': 'PPS-L1C',
                       'type': 'NETCDF',
                       'data_processing_level': '1c'}

        with patch.object(my_fake_publisher, 'send') as mock:
            publish_l1c(my_fake_publisher, pub_message, ['/1c/nc/0deg'])

        mock.assert_called_once()
        mock.assert_called_once_with("some pytroll message")


class TestL1cProcessing(unittest.TestCase):
    """Test the L1c processing module."""

    def test_check_service_name_okay(self):

        check_service_name_okay('seviri-l1c')
        check_service_name_okay('viirs-l1c')
        check_service_name_okay('avhrr-l1c')
        check_service_name_okay('modis-l1c')

        self.assertRaises(ServiceNameNotSupported, check_service_name_okay, 'seviri')

        with pytest.raises(ServiceNameNotSupported) as exec_info:
            check_service_name_okay('avhrr')

        exception_raised = exec_info.value
        self.assertTrue('Service name avhrr is not yet supported' == str(exception_raised))

    def test_check_message_okay_message_ok(self):

        input_msg = Message.decode(rawstr=TEST_INPUT_MSG)

        result = check_message_okay(input_msg)
        self.assertTrue(result)

    def test_check_message_okay_message_has_no_dataset(self):
        """Test that message is not okay if it is not a dataset."""

        input_msg = Message.decode(rawstr=TEST_INPUT_MSG_NO_DATASET)
        result = check_message_okay(input_msg)
        self.assertTrue(result is False)

    def test_check_message_okay_message_has_no_platform_name(self):
        """Test that message is not okay if it is not a dataset."""

        input_msg = Message.decode(rawstr=TEST_INPUT_MSG_NO_PLATFORM_NAME)
        result = check_message_okay(input_msg)
        self.assertTrue(result is False)

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	# Copyright (c) 2021 Adam.Dybbroe
5
6	# Author(s):
7
8	# Adam.Dybbroe <a000680@c21856.ad.smhi.se>
9
10	# This program is free software: you can redistribute it and/or modify
11	# it under the terms of the GNU General Public License as published by
12	# the Free Software Foundation, either version 3 of the License, or
13	# (at your option) any later version.
14
15	# This program is distributed in the hope that it will be useful,
16	# but WITHOUT ANY WARRANTY; without even the implied warranty of
17	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18	# GNU General Public License for more details.
19
20	# You should have received a copy of the GNU General Public License
21	# along with this program. If not, see <http://www.gnu.org/licenses/>.
22
23	"""Unit testing the level-1c runner code	4✔
24	"""
25
26	import pytest	4✔
27	from unittest.mock import patch	4✔
28	import unittest	4✔
29	from posttroll.message import Message	4✔
30	from datetime import datetime	4✔
31
32	from nwcsafpps_runner.message_utils import publish_l1c, prepare_l1c_message	4✔
33	from nwcsafpps_runner.l1c_processing import check_message_okay	4✔
34	from nwcsafpps_runner.l1c_processing import check_service_name_okay	4✔
35	# from nwcsafpps_runner.l1c_processing import L1cProcessor
36	from nwcsafpps_runner.l1c_processing import ServiceNameNotSupported	4✔
37
38
39	class MyFakePublisher(object):	4✔
40
41	def __init__(self):	4✔
42	pass	4✔
43
44	def send(self, message):	4✔
NEW 45	pass	×
46
47
48	TEST_INPUT_MSG = """pytroll://1b/hrit/0deg dataset safusr.u@lxserv1043.smhi.se 2021-05-18T14:28:54.154172 v1.01 application/json {"data_type": "MSG4", "orig_platform_name": "MSG4", "start_time": "2021-05-18T14:15:00", "variant": "0DEG", "series": "MSG4", "platform_name": "Meteosat-11", "channel": "", "nominal_time": "2021-05-18T14:15:00", "compressed": "", "origin": "172.18.0.248:9093", "dataset": [{"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__", "uid": "H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__"}, {"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__", "uid": "H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__"}], "sensor": ["seviri"]}"""	4✔
49
50	TEST_INPUT_MSG_NO_DATASET = """pytroll://1b/hrit/0deg file safusr.u@lxserv1043.smhi.se 2021-05-18T14:28:54.154172 v1.01 application/json {"data_type": "MSG4", "orig_platform_name": "MSG4", "start_time": "2021-05-18T14:15:00", "variant": "0DEG", "series": "MSG4", "platform_name": "Meteosat-11", "channel": "", "nominal_time": "2021-05-18T14:15:00", "compressed": "", "origin": "172.18.0.248:9093", "file": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__", "sensor": ["seviri"]}"""	4✔
51
52
53	TEST_INPUT_MSG_NO_PLATFORM_NAME = """pytroll://1b/hrit/0deg dataset safusr.u@lxserv1043.smhi.se 2021-05-18T14:28:54.154172 v1.01 application/json {"data_type": "MSG4", "orig_platform_name": "MSG4", "start_time": "2021-05-18T14:15:00", "variant": "0DEG", "series": "MSG4", "channel": "", "nominal_time": "2021-05-18T14:15:00", "compressed": "", "origin": "172.18.0.248:9093", "dataset": [{"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__", "uid": "H-000-MSG4__-MSG4________-_________-PRO______-202105181415-__"}, {"uri": "/san1/geo_in/0deg/H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__", "uid": "H-000-MSG4__-MSG4________-HRV______-000001___-202105181415-__"}], "sensor": ["seviri"]}"""	4✔
54
55
56	class TestPublishMessage(unittest.TestCase):	4✔
57
58	@patch('nwcsafpps_runner.message_utils.socket.gethostname')	4✔
59	def test_create_publish_message(self, gethostname):	3✔
60	"""Test the creation of the publish message."""
61
62	gethostname.return_value = "my_local_server"	4✔
63	my_fake_level1c_file = '/my/level1c/file/path/level1c.nc'	4✔
64	input_msg = Message.decode(rawstr=TEST_INPUT_MSG)	4✔
65
66	result = prepare_l1c_message(my_fake_level1c_file, input_msg.data, orbit=99999)	4✔
67
68	expected = {'data_type': 'MSG4', 'orig_platform_name': 'MSG4',	4✔
69	'start_time': datetime(2021, 5, 18, 14, 15),
70	'variant': '0DEG', 'series': 'MSG4',
71	'platform_name': 'Meteosat-11',
72	'channel': '',
73	'nominal_time': datetime(2021, 5, 18, 14, 15),
74	'compressed': '',
75	'origin': '172.18.0.248:9093', 'sensor': ['seviri'],
76	'uri': 'ssh://my_local_server/my/level1c/file/path/level1c.nc',
77	'uid': 'level1c.nc',
78	'format': 'PPS-L1C',
79	'type': 'NETCDF',
80	'data_processing_level': '1c'}
81
82	self.assertDictEqual(result, expected)	4✔
83
84	@patch('nwcsafpps_runner.message_utils.Message.encode')	4✔
85	def test_publish_messages(self, mock_message):	3✔
86	"""Test the sending the messages."""
87
88	my_fake_publisher = MyFakePublisher()	4✔
89	mock_message.return_value = "some pytroll message"	4✔
90
91	pub_message = {'data_type': 'MSG4', 'orig_platform_name': 'MSG4',	4✔
92	'start_time': datetime(2021, 5, 18, 14, 15),
93	'variant': '0DEG', 'series': 'MSG4',
94	'platform_name': 'Meteosat-11',
95	'channel': '',
96	'nominal_time': datetime(2021, 5, 18, 14, 15),
97	'compressed': '',
98	'origin': '172.18.0.248:9093', 'sensor': ['seviri'],
99	'uri': 'ssh://my_local_server/my/level1c/file/path/level1c.nc',
100	'uid': 'level1c.nc',
101	'format': 'PPS-L1C',
102	'type': 'NETCDF',
103	'data_processing_level': '1c'}
104
105	with patch.object(my_fake_publisher, 'send') as mock:	4✔
106	publish_l1c(my_fake_publisher, pub_message, ['/1c/nc/0deg'])	4✔
107
108	mock.assert_called_once()	4✔
109	mock.assert_called_once_with("some pytroll message")	4✔
110
111
112	class TestL1cProcessing(unittest.TestCase):	4✔
113	"""Test the L1c processing module."""
114
115	def test_check_service_name_okay(self):	4✔
116
117	check_service_name_okay('seviri-l1c')	4✔
118	check_service_name_okay('viirs-l1c')	4✔
119	check_service_name_okay('avhrr-l1c')	4✔
120	check_service_name_okay('modis-l1c')	4✔
121
122	self.assertRaises(ServiceNameNotSupported, check_service_name_okay, 'seviri')	4✔
123
124	with pytest.raises(ServiceNameNotSupported) as exec_info:	4✔
125	check_service_name_okay('avhrr')	4✔
126
127	exception_raised = exec_info.value	4✔
128	self.assertTrue('Service name avhrr is not yet supported' == str(exception_raised))	4✔
129
130	def test_check_message_okay_message_ok(self):	4✔
131
132	input_msg = Message.decode(rawstr=TEST_INPUT_MSG)	4✔
133
134	result = check_message_okay(input_msg)	4✔
135	self.assertTrue(result)	4✔
136
137	def test_check_message_okay_message_has_no_dataset(self):	4✔
138	"""Test that message is not okay if it is not a dataset."""
139
140	input_msg = Message.decode(rawstr=TEST_INPUT_MSG_NO_DATASET)	4✔
141	result = check_message_okay(input_msg)	4✔
142	self.assertTrue(result is False)	4✔
143
144	def test_check_message_okay_message_has_no_platform_name(self):	4✔
145	"""Test that message is not okay if it is not a dataset."""
146
147	input_msg = Message.decode(rawstr=TEST_INPUT_MSG_NO_PLATFORM_NAME)	4✔
148	result = check_message_okay(input_msg)	4✔
149	self.assertTrue(result is False)	4✔

pytroll / pytroll-pps-runner / 863878245

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous