backport: PR8720 and PR8754 (#9361)

* fix(agw): Fixing pylint tests for open function missing encoding param (#8720)

Signed-off-by: Alex Rodriguez <ardzoht@gmail.com>

* fix(agw): Fix bootstrap failure due to bad Python open args (#8754)

Co-authored-by: Alex Rodriguez <alexrod@fb.com>
Co-authored-by: Hunter Gatewood <hgatewood@gmail.com>

lte/gateway/python/magma/enodebd/enodeb_status.py

@@ -555,7 +555,7 @@ def _get_cached_gps_coords() -> Tuple[str, str]:
 
 def _read_gps_coords_from_file():
     try:
-        with open(CACHED_GPS_COORD_FILE_PATH) as f:
+        with open(CACHED_GPS_COORD_FILE_PATH, encoding="utf-8") as f:
             lines = f.readlines()
             if len(lines) != 2:
                 logger.warning(

lte/gateway/python/magma/health/health_service.py

@@ -65,7 +65,7 @@ def get_subscriber_table(self):
         return table.entries
 
     def get_registration_success_rate(self, mme_log_file):
-        with open(mme_log_file, 'r') as f:
+        with open(mme_log_file, 'r', encoding="utf-8") as f:
             log = f.read()
 
         return RegistrationSuccessRate(

lte/gateway/python/magma/pipelined/qos/qos_tc_impl.py

@@ -84,7 +84,7 @@ def init_qdisc(intf: str, show_error=False, enable_pyroute2=False) -> int:
         qid_hex = hex(ROOT_QID)
         fn = "/sys/class/net/{intf}/speed".format(intf=intf)
         try:
-            with open(fn) as f:
+            with open(fn, encoding="utf-8") as f:
                 speed = f.read().strip()
         except OSError:
             LOG.error('unable to read speed from %s defaulting to %s', fn, speed)

lte/gateway/python/magma/subscriberdb/tests/protocols/diameter/avp_tests.py

@@ -255,14 +255,14 @@ def test_octet_strings(self):
 
         # Unicode strings won't load
         with self.assertRaises(CodecException):
-            avp.OctetStringAVP(0, u'hello')
+            avp.OctetStringAVP(0, 'hello')
 
     def test_unicode_strings(self):
         """
         Tests we can encode and decode unicode strings
         """
         self._compare_avp(
-            avp.UTF8StringAVP(1, u'\u0123\u0490'),
+            avp.UTF8StringAVP(1, '\u0123\u0490'),
             memoryview(b'\x00\x00\x00\x01\x00\x00\x00\x0c\xc4\xa3\xd2\x90'),
         )
 

lte/gateway/python/magma/tests/pylint_tests.py

@@ -38,6 +38,7 @@ def test_pylint(self):
                 'fixme',  # allow todos
                 'unnecessary-pass',  # triggers when pass is ok
                 'raise-missing-from',
+                'redundant-u-string-prefix',
             ],
             show_categories=["warning", "error", "fatal"],
         )

orc8r/gateway/python/magma/common/health/health_service.py

@@ -99,7 +99,7 @@ def get_error_summary(self, service_names):
                 'Or execute the command with sudo '
                 'permissions: `venvsudo`'.format(syslog_path),
             )
-        with open(syslog_path, 'r') as f:
+        with open(syslog_path, 'r', encoding='utf-8') as f:
             for line in f:
                 for service_name in service_names:
                     if service_name not in line:

orc8r/gateway/python/magma/common/service_registry.py

@@ -254,14 +254,17 @@ def get_ssl_creds():
     """
     proxy_config = ServiceRegistry.get_proxy_config()
     try:
-        rootca = open(proxy_config['rootca_cert'], 'rb').read()
-        cert = open(proxy_config['gateway_cert']).read().encode()
-        key = open(proxy_config['gateway_key']).read().encode()
-        ssl_creds = grpc.ssl_channel_credentials(
-            root_certificates=rootca,
-            certificate_chain=cert,
-            private_key=key,
-        )
+        with open(proxy_config['rootca_cert'], 'rb') as rootca_f:
+            with open(proxy_config['gateway_cert'], encoding="utf-8") as cert_f:
+                with open(proxy_config['gateway_key'], encoding="utf-8") as key_f:
+                    rootca = rootca_f.read()
+                    cert = cert_f.read().encode()
+                    key = key_f.read().encode()
+                    ssl_creds = grpc.ssl_channel_credentials(
+                        root_certificates=rootca,
+                        certificate_chain=cert,
+                        private_key=key,
+                    )
     except FileNotFoundError as exp:
         raise ValueError("SSL cert not found: %s" % exp)
     return ssl_creds

orc8r/gateway/python/magma/configuration/mconfig_managers.py

@@ -153,7 +153,7 @@ class MconfigManagerImpl(MconfigManager[GatewayConfigs]):
     def load_mconfig(self) -> GatewayConfigs:
         cfg_file_name = self._get_mconfig_file_path()
         try:
-            with open(cfg_file_name, 'r') as cfg_file:
+            with open(cfg_file_name, 'r', encoding='utf-8') as cfg_file:
                 mconfig_str = cfg_file.read()
             return self.deserialize_mconfig(mconfig_str)
         except (OSError, json.JSONDecodeError, json_format.ParseError) as e:
@@ -174,7 +174,7 @@ def load_service_mconfig(
 
     def load_service_mconfig_as_json(self, service_name) -> Any:
         cfg_file_name = self._get_mconfig_file_path()
-        with open(cfg_file_name, 'r') as f:
+        with open(cfg_file_name, 'r', encoding='utf-8') as f:
             json_mconfig = json.load(f)
             service_configs = json_mconfig.get('configsByKey', {})
             service_configs.update(json_mconfig.get('configs_by_key', {}))

orc8r/gateway/python/magma/configuration/service_configs.py

@@ -55,7 +55,7 @@ def save_override_config(service_name: str, cfg: Any):
     """
     override_file_name = _override_file_name(service_name)
     os.makedirs(CONFIG_OVERRIDE_DIR, exist_ok=True)
-    with open(override_file_name, 'w') as override_file:
+    with open(override_file_name, 'w', encoding='utf-8') as override_file:
         yaml.dump(cfg, override_file, default_flow_style=False)
 
 
@@ -144,7 +144,7 @@ def _load_yaml_file(file_name: str) -> Any:
     """
 
     try:
-        with open(file_name, 'r') as stream:
+        with open(file_name, 'r', encoding='utf-8') as stream:
             data = yaml.safe_load(stream)
             return data
     except (OSError, yaml.YAMLError) as e:

orc8r/gateway/python/magma/magmad/metrics.py

@@ -233,7 +233,7 @@ def monitor_unattended_upgrade_status():
         status = 0
         auto_upgrade_file_name = '/etc/apt/apt.conf.d/20auto-upgrades'
         if os.path.isfile(auto_upgrade_file_name):
-            with open(auto_upgrade_file_name) as auto_upgrade_file:
+            with open(auto_upgrade_file_name, encoding='utf-8') as auto_upgrade_file:
                 for line in auto_upgrade_file:
                     package_name, flag = line.strip().strip(';').split()
                     if package_name == "APT::Periodic::Unattended-Upgrade":

orc8r/gateway/python/magma/magmad/upgrade/docker_upgrader.py

@@ -69,7 +69,7 @@ async def get_versions(self) -> VersionInfo:
         """ Returns the current version by parsing the IMAGE_VERSION in the
         .env file
         """
-        with open('/var/opt/magma/docker/.env', 'r') as env:
+        with open('/var/opt/magma/docker/.env', 'r', encoding='utf-8') as env:
             for line in env:
                 if line.startswith("IMAGE_VERSION="):
                     current_version = line.split("=")[1].strip()

orc8r/gateway/python/magma/tests/pylint_tests.py

@@ -38,6 +38,7 @@ def test_pylint(self):
                 'fixme',  # allow todos
                 'unnecessary-pass',  # triggers when pass is ok
                 'raise-missing-from',
+                'redundant-u-string-prefix',
             ],
             show_categories=["warning", "error", "fatal"],
         )