ruff: Fix SIM115 Use context handler for opening files.

Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg 2023-01-17 22:06:51 -05:00 committed by Tim Abbott
parent c6b6004432
commit 6303ebfc2f
4 changed files with 71 additions and 69 deletions

View File

@ -2,7 +2,7 @@ import os
import subprocess
import sys
import time
from contextlib import contextmanager
from contextlib import ExitStack, contextmanager
from typing import Iterator, Optional
# Verify the Zulip venv is available.
@ -58,13 +58,14 @@ def test_server_running(
log_file: Optional[str] = None,
dots: bool = False,
) -> Iterator[None]:
with ExitStack() as stack:
log = sys.stdout
if log_file:
if os.path.exists(log_file) and os.path.getsize(log_file) < 100000:
log = open(log_file, "a")
log = stack.enter_context(open(log_file, "a"))
log.write("\n\n")
else:
log = open(log_file, "w")
log = stack.enter_context(open(log_file, "w"))
set_up_django(external_host)

View File

@ -314,7 +314,8 @@ def main() -> None:
if not file_name[0].isalnum() or not file_name.endswith(".py"):
continue
filepath = os.path.join(root, file_name)
for line in open(filepath):
with open(filepath) as f:
for line in f:
if search_key not in line:
continue
new_suite = filepath.replace(".py", ".") + suite

View File

@ -12,6 +12,24 @@ def error(*args: Any) -> None:
raise Exception("We cannot enqueue because settings.USING_RABBITMQ is False.")
def enqueue_file(queue_name: str, f: IO[str]) -> None:
for line in f:
line = line.strip()
try:
payload = line.split("\t")[1]
except IndexError:
payload = line
print(f"Queueing to queue {queue_name}: {payload}")
# Verify that payload is valid json.
data = orjson.loads(payload)
# This is designed to use the `error` method rather than
# the call_consume_in_tests flow.
queue_json_publish(queue_name, data, error)
class Command(BaseCommand):
help = """Read JSON lines from a file and enqueue them to a worker queue.
@ -36,26 +54,7 @@ You can use "-" to represent stdin.
file_name = options["file_name"]
if file_name == "-":
f: IO[str] = sys.stdin
enqueue_file(queue_name, sys.stdin)
else:
f = open(file_name)
while True:
line = f.readline()
if not line:
break
line = line.strip()
try:
payload = line.split("\t")[1]
except IndexError:
payload = line
print(f"Queueing to queue {queue_name}: {payload}")
# Verify that payload is valid json.
data = orjson.loads(payload)
# This is designed to use the `error` method rather than
# the call_consume_in_tests flow.
queue_json_publish(queue_name, data, error)
with open(file_name) as f:
enqueue_file(queue_name, f)

View File

@ -41,7 +41,8 @@ def test_generated_curl_examples_for_success(client: Client) -> None:
rest_endpoints_path = os.path.join(
settings.DEPLOY_ROOT, "templates/zerver/api/include/rest-endpoints.md"
)
rest_endpoints_raw = open(rest_endpoints_path).read()
with open(rest_endpoints_path) as f:
rest_endpoints_raw = f.read()
ENDPOINT_REGEXP = re.compile(r"/api/\s*(.*?)\)")
endpoint_list = sorted(set(re.findall(ENDPOINT_REGEXP, rest_endpoints_raw)))
@ -51,7 +52,7 @@ def test_generated_curl_examples_for_success(client: Client) -> None:
curl_commands_to_test = []
if os.path.exists(file_name):
f = open(file_name)
with open(file_name) as f:
for line in f:
# A typical example from the Markdown source looks like this:
# {generate_code_example(curl)|...|...}