Merge branch 'dev' into vas

Author: congnghiahieu

src/schemathesis/cli/output/default.py

@@ -1,43 +1,44 @@
 from __future__ import annotations
+
 import base64
 import os
 import platform
 import shutil
 import textwrap
 import time
+from importlib import metadata
 from itertools import groupby
 from queue import Queue
 from typing import Any, Generator, cast
 
 import click
-from importlib import metadata
 
 from ... import service
 from ...code_samples import CodeSampleStyle
 from ...constants import (
  DISCORD_LINK,
+ FALSE_VALUES,
  FLAKY_FAILURE_MESSAGE,
+ GITHUB_APP_LINK,
+ ISSUE_TRACKER_URL,
  REPORT_SUGGESTION_ENV_VAR,
  SCHEMATHESIS_TEST_CASE_HEADER,
  SCHEMATHESIS_VERSION,
- FALSE_VALUES,
- ISSUE_TRACKER_URL,
- GITHUB_APP_LINK,
 )
 from ...exceptions import RuntimeErrorType, prepare_response_payload
 from ...experimental import GLOBAL_EXPERIMENTS
 from ...models import Status
 from ...runner import events
 from ...runner.events import InternalErrorType, SchemaErrorType
 from ...runner.serialization import (
+ SerializedCheck,
  SerializedError,
  SerializedTestResult,
  deduplicate_failures,
- SerializedCheck,
 )
+from ...specs.openapi._vas import logger
 from ..context import ExecutionContext, FileReportContext, ServiceReportContext
 from ..handlers import EventHandler
-from ...specs.openapi._vas import logger
 
 SPINNER_REPETITION_NUMBER = 10
 
@@ -365,7 +366,8 @@ def _by_unique_code_sample(
  check: SerializedCheck, code_sample_style: CodeSampleStyle
 ) -> str:
  request_body = (
- base64.b64decode(check.example.body).decode()
+ # Ignore decoding errors caused by invalid base64
+ base64.b64decode(check.example.body).decode('utf-8', 'ignore')
  if check.example.body is not None
  else None
  )
@@ -531,7 +533,7 @@ def display_service_unauthorized(hostname: str) -> None:
 
 def display_service_error(event: service.Error, message_prefix: str = "") -> None:
  """Show information about an error during communication with Schemathesis.io."""
- from requests import RequestException, HTTPError, Response
+ from requests import HTTPError, RequestException, Response
 
  if isinstance(event.exception, HTTPError):
  response = cast(Response, event.exception.response)

src/schemathesis/models.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import base64
 import datetime
 import inspect
 import textwrap
@@ -9,8 +10,7 @@
 from enum import Enum
 from functools import lru_cache, partial
 from itertools import chain, count
-from logging import Logger, LogRecord
-from math import e
+from logging import LogRecord
 from typing import (
  TYPE_CHECKING,
  Any,
@@ -127,6 +127,7 @@ def prepare_request_data(kwargs: dict[str, Any]) -> PreparedRequestData:
  if key in get_request_signature().parameters
  }
  request = requests.Request(**kwargs).prepare()
+ # logger.debug("in prepare request data ", request.body)
  return PreparedRequestData(
  method=str(request.method),
  url=str(request.url),
@@ -351,15 +352,23 @@ def as_requests_kwargs(
 
  if isinstance(self.body, dict):
  # Create a new body without meta-data
- new_body = {
+ body_without_prefixed_field = {
  key: value
  for key, value in self.body.items()
  if not key.startswith(VAS_KEY_PREFIX)
  }
- self.metadata = {
- f"{key[len(VAS_KEY_PREFIX) + 1:]}": value
- for key, value in self.body.items()
- if key.startswith(VAS_KEY_PREFIX)
+ if self.metadata == {}:
+ self.metadata = {
+ f"{key[len(VAS_KEY_PREFIX) + 1:]}": value
+ for key, value in self.body.items()
+ if key.startswith(VAS_KEY_PREFIX)
+ }
+ new_body = {
+ **body_without_prefixed_field,
+ **{
+ k: obj["image_name"]
+ for k, obj in self.metadata.items()
+ }
  }
  self.body = new_body
 
@@ -409,7 +418,8 @@ def as_requests_kwargs(
  "deps/schemathesis/src/schemathesis/models.py -> new extra: %s", new_extra
  )
 
- additional_headers = extra.pop("headers", None)
+ additional_headers = new_extra.pop("headers", None)
+ # additional_headers = extra.pop("headers", None)
  if additional_headers:
  # Additional headers, needed for the serializer
  for key, value in additional_headers.items():
@@ -421,6 +431,7 @@ def as_requests_kwargs(
  "headers": final_headers,
  "params": self.query,
  **new_extra,
+ # **extra,
  }
 
  def call(
@@ -863,6 +874,7 @@ def as_strategy(
  ) -> st.SearchStrategy:
  """Turn this API operation into a Hypothesis strategy."""
 
+ logger.debug("deps/schemathesis/src/schemathesis/models.py: as_strategy -> kwargs: %s", kwargs)
  # Cai nay quan trong cuc ki
  strategy = self.schema.get_case_strategy(
  self,
@@ -1054,12 +1066,17 @@ def from_case(cls, case: Case, session: requests.Session) -> Request:
  base_url = case.get_full_base_url()
  kwargs = case.as_requests_kwargs(base_url)
  request = requests.Request(**kwargs)
+ # logger.debug("deps/schemathesis/src/schemathesis/models.py: from_case in Request -> request: %s", request)
  prepared = session.prepare_request(request)
  return cls.from_prepared_request(prepared)
 
  @classmethod
  def from_prepared_request(cls, prepared: requests.PreparedRequest) -> Request:
  """A prepared request version is already stored in `requests.Response`."""
+ logger.debug(
+ "deps/schemathesis/src/schemathesis/models.py: from_prepared_request in Request -> body: %s",
+ prepared.body,
+ )
  body = prepared.body
  if isinstance(body, str):
  # can be a string for `application/x-www-form-urlencoded`
@@ -1172,6 +1189,15 @@ def from_requests(
  status: Status,
  checks: list[Check],
  ) -> Interaction:
+
+ logger.debug(
+ "deps/schemathesis/src/schemathesis/models.py: from_requests in Interaction -> response.request.body: %s",
+ response.request.body,
+ )
+ logger.debug(
+ "deps/schemathesis/src/schemathesis/models.py: from_requests in Interaction -> case.body: %s",
+ case.body,
+ )
  return cls(
  request=Request.from_prepared_request(response.request),
  response=Response.from_requests(response),
@@ -1198,6 +1224,7 @@ def from_wsgi(
  session = requests.Session()
  session.headers.update(headers)
  return cls(
+ case=case,
  request=Request.from_case(case, session),
  response=Response.from_wsgi(response, elapsed),
  status=status,

src/schemathesis/runner/impl/core.py

@@ -54,13 +54,14 @@
 )
 from ...runner import events
 from ...schemas import BaseSchema
+from ...specs.openapi._vas import logger
 from ...stateful import Feedback, Stateful
 from ...targets import Target, TargetContext
 from ...types import RawAuth, RequestCert
 from ...utils import capture_hypothesis_output
 from ..override import CaseOverride
 from ..serialization import SerializedTestResult
-from ...specs.openapi._vas import logger
+
 if TYPE_CHECKING:
  from ...transports.responses import GenericResponse, WSGIResponse
 
@@ -791,6 +792,43 @@ def _network_test(
  run_targets(targets, context)
  status = Status.success
  try:
+ # logger.debug("kwargs in core.py: %s", kwargs)
+ requests_kwargs = case.as_requests_kwargs(
+ base_url=case.get_full_base_url(), headers=headers
+ )
+ if 'files' in requests_kwargs:
+ # existing_tuple = requests_kwargs['files'][0]
+ #logger.debug("existing tuple in kwargs in core.py: %s", existing_tuple)
+
+ if case.metadata != {}:
+ for i, (key, value) in enumerate(requests_kwargs['files']):
+ # Check if the key is in metadata
+ if key in case.metadata:
+ # Check if the current value is a tuple
+ if isinstance(value, tuple):
+ # Create a new tuple with the updated image name
+ new_tuple = (value[0], case.metadata[key]["image_name"], *value[2:])
+ # Update the list with the new tuple
+ requests_kwargs['files'][i] = (key, new_tuple)
+ else:
+ # If not a tuple, directly update with the new image name
+ requests_kwargs['files'][i] = (key, case.metadata[key]["image_name"])
+ #logger.debug("Updated file in kwargs: %s", requests_kwargs['files'][i])
+ # requests_kwargs['files'][0][1] = case.metadata['image_name']
+ #logger.debug("files in kwargs in core.py: %s", requests_kwargs['files'][0][1])
+ #logger.debug("request_kargs in core.py: %s", requests_kwargs)
+ ### Create new request but reserve old content length
+ old_content_length = response.request.headers.get('Content-Length', '0')
+ #logger.debug("old request content length in core.py: %s", old_content_length)
+ request = requests.Request(**requests_kwargs).prepare()
+ #logger.debug("request content length in core.py: %s", request.headers.get('Content-Length'))
+ request.headers['Content-Length'] = old_content_length # type: ignore
+
+ ### Create new request with new content length
+ # request = requests.Request(**requests_kwargs).prepare()
+ #logger.debug("new request content length in core.py: %s", request.headers.get('Content-Length'))
+ response.request = request
+ #logger.debug("case in core.py: %s", case.body)
  run_checks(
  case=case,
  checks=checks,
@@ -804,11 +842,12 @@ def _network_test(
  status = Status.failure
  raise
  finally:
- # if feedback.stateful == Stateful.links:
- # logger.debug("stateful")
+ if feedback.stateful == Stateful.links:
+  logger.debug("Stateful")
  # do something
 
  feedback.add_test_case(case, response)
+ # logger.debug("feedback in core.py: ", feedback)
  if store_interactions:
  # Lưu lại interaction
  result.store_requests_response(case, response, status, check_results)

src/schemathesis/runner/serialization.py

@@ -3,8 +3,8 @@
 They all consist of primitive types and don't have references to schemas, app, etc.
 """
 from __future__ import annotations
-
 import logging
+from ..specs.openapi._vas import logger
 import re
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, cast
@@ -367,6 +367,7 @@ class SerializedInteraction:
 
  @classmethod
  def from_interaction(cls, interaction: Interaction) -> SerializedInteraction:
+ # logger.debug("deps/schemathesis/src/schemathesis/runner/serialization.py: serializing interaction")
  return cls(
  request=interaction.request,
  response=interaction.response,

src/schemathesis/serializers.py

@@ -81,9 +81,6 @@ class Serializer(Protocol):
  """
 
  def as_requests(self, context: SerializerContext, payload: Any) -> dict[str, Any]:
- logger.debug(
- "deps/schemathesis/src/schemathesis/serializers.py: Serializing body"
- )
  raise NotImplementedError
 
  def as_werkzeug(self, context: SerializerContext, payload: Any) -> dict[str, Any]:
@@ -256,6 +253,7 @@ def as_requests(self, context: SerializerContext, value: Any) -> dict[str, Any]:
  value = fast_deepcopy(value)
  multipart = _prepare_form_data(value)
  files, data = context.case.operation.prepare_multipart(multipart)
+ # logger.debug("deps/schemathesis/src/schemathesis/serializers.py: files: %s", files)
  return {"files": files, "data": data}
  # Uncommon schema. For example - `{"type": "string"}`
  boundary = choose_boundary()

src/schemathesis/specs/openapi/_vas.py

@@ -3,6 +3,7 @@
 import sys
 from functools import partial
 
+from dotenv import dotenv_values as get_dotenv_values
 from hypothesis import strategies as st
 
 # from faker import Faker
@@ -30,7 +31,9 @@
 # from faker_file.providers.zip_file import ZipFileProvider
 
 
-DEV = False
+dotenv_values = get_dotenv_values(".env")
+PYTHON_ENV = dotenv_values.get("PYTHON_ENV", None)
+DEV = PYTHON_ENV != "PROD"
 
 # FAKER = Faker()
 
@@ -97,7 +100,6 @@
  # "binary": get_file_strategy().map(Binary),
  # "byte": get_file_strategy().map(lambda x: b64encode(x).decode()),
 }
-
 CURRENT_LEVEL = logging.DEBUG if DEV else logging.INFO
 CURRENT_FORMAT = (
  "%(asctime)s %(filename)s:%(lineno)d:%(funcName)s %(levelname)s:%(message)s"

src/schemathesis/specs/openapi/schemas.py

@@ -1115,7 +1115,6 @@ def get_response_schema(
  scopes, definition = self.resolver.resolve_in_scope(
  fast_deepcopy(definition), scope
  )
- logger.debug("deps/schemathesis/src/schemathesis/specs/openapi/schemas.py: get_response_schema -> ")
  options = iter(definition.get("content", {}).values())
  option = next(options, None)
  # "schema" is an optional key in the `MediaType` object