python: Catch specific exceptions from subprocess.

Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg 2020-10-08 18:32:00 -07:00 committed by Tim Abbott
parent aabef3d9be
commit 17ac17286c
6 changed files with 9 additions and 6 deletions

View File

@ -2,6 +2,7 @@ import base64
import os
import re
import shutil
import subprocess
import tempfile
import urllib
from contextlib import contextmanager
@ -970,7 +971,7 @@ Output:
'''
with \
self.settings(ERROR_BOT=None), \
mock.patch('zerver.lib.markdown.timeout', side_effect=KeyError('foo')), \
mock.patch('zerver.lib.markdown.timeout', side_effect=subprocess.CalledProcessError(1, [])), \
mock.patch('zerver.lib.markdown.markdown_logger'):
yield

View File

@ -26,7 +26,7 @@ def try_git_describe() -> Optional[str]:
stderr=subprocess.PIPE,
cwd=os.path.join(os.path.dirname(__file__), '..'),
).strip().decode('utf-8')
except Exception: # nocoverage
except (FileNotFoundError, subprocess.CalledProcessError): # nocoverage
return None
def add_request_metadata(report: Dict[str, Any], request: HttpRequest) -> None:

View File

@ -1,3 +1,4 @@
import subprocess
from typing import Any, Callable, Dict, Iterable, List, Tuple
from unittest import mock
@ -148,7 +149,7 @@ class TestReport(ZulipTestCase):
subprocess_mock = mock.patch(
'zerver.views.report.subprocess.check_output',
side_effect=KeyError('foo'),
side_effect=subprocess.CalledProcessError(1, []),
)
with mock_queue_publish('zerver.views.report.queue_json_publish') as m, subprocess_mock:
result = self.client_post("/json/report/error", params)

View File

@ -1,3 +1,4 @@
import subprocess
from typing import Any
from unittest.mock import patch
@ -51,7 +52,7 @@ class ZephyrTest(ZulipTestCase):
with \
ccache_mock(return_value=b'1234'), \
ssh_mock(side_effect=KeyError('foo')), \
ssh_mock(side_effect=subprocess.CalledProcessError(1, [])), \
logging_mock() as log:
result = post("zephyr", cred=cred)

View File

@ -108,7 +108,7 @@ def report_error(request: HttpRequest, user_profile: UserProfile, message: str=R
["git", "show", "-s", "--oneline"],
universal_newlines=True,
)
except Exception:
except (FileNotFoundError, subprocess.CalledProcessError):
version = None
# Get the IP address of the request

View File

@ -66,7 +66,7 @@ def webathena_kerberos_login(request: HttpRequest, user_profile: UserProfile,
]
subprocess.check_call(["ssh", settings.PERSONAL_ZMIRROR_SERVER, "--",
" ".join(map(shlex.quote, command))])
except Exception:
except subprocess.CalledProcessError:
logging.exception("Error updating the user's ccache", stack_info=True)
return json_error(_("We were unable to setup mirroring for you"))