19482063855

Committed 18 Nov 2025 09:34PM UTC coverage: 93.533% (-0.5%) from 94.065%

Build # 19482063855

Build Type

Pull #1075

github

Committed by

dchhabda

Commit Message

modified boto2-3 migration

Pull Request Pull Request #1075: RDISCROWD-8392: deprecate old boto. use boto3 only (Updated)

Run Details

10 of 19 new or added lines in 3 files covered. (52.63%)

87 existing lines in 5 files now uncovered.

17703 of 18927 relevant lines covered (93.53%)

0.94 hits per line

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

75.47

/pybossa/cloud_store_api/connection.py

from copy import deepcopy
import ssl
import sys
import time

from flask import current_app

from werkzeug.exceptions import BadRequest
from boto3.session import Session
from botocore.client import Config
from pybossa.cloud_store_api.base_conn import BaseConnection
from os import environ


def check_store(store):
    if not store:
        return

    store_type = current_app.config.get("S3_CONN_TYPE")
    store_type_v2 = current_app.config.get("S3_CONN_TYPE_V2")
    if store not in [store_type, store_type_v2]:
        raise BadRequest(f"Unsupported store type {store}")

def create_connection(**kwargs):
    # TODO: remove later
    v2_access = environ.get("AWS_V2_ACCESS_KEY_ID")
    v2_secret = environ.get("AWS_V2_SECRET_ACCESS_KEY")
    if v2_access and v2_secret:
        masked_v2_secret = f"{v2_secret[:3]}{'x'*(len(v2_secret)-6)}{v2_secret[-3:]}"
        current_app.logger.info("v2_access %s, v2_secret %s", v2_access, masked_v2_secret)
    else:
        current_app.logger.info("v2_access, v2_secret not found")

    if kwargs.get("aws_secret_access_key"):
        masked_kwargs = {k:v for k, v in kwargs.items()}
        secret = kwargs["aws_secret_access_key"]
        masked_kwargs["aws_secret_access_key"] = f"{secret[:3]}{'x'*(len(secret)-6)}{secret[-3:]}"
        current_app.logger.info(f"create_connection kwargs: %s", str(masked_kwargs))
    else:
        current_app.logger.info(f"create_connection kwargs: %s", str(kwargs))

    store = kwargs.pop("store", None)
    check_store(store)
    store_type_v2 = current_app.config.get("S3_CONN_TYPE_V2")
    if store and store == store_type_v2:
        current_app.logger.info("Calling CustomConnectionV2")
        return CustomConnectionV2(
            aws_access_key_id=kwargs.get("aws_access_key_id"),
            aws_secret_access_key=kwargs.get("aws_secret_access_key"),
            endpoint=kwargs.get("endpoint"),
            cert=kwargs.get("cert", False),
            proxy_url=kwargs.get("proxy_url")
        )

    current_app.logger.info("Calling CustomConnection")
    conn = CustomConnection(**kwargs)
    return conn


class CustomConnection(BaseConnection):

    def __init__(self, *args, **kwargs):
        super().__init__()  # super(CustomConnection, self).__init__(*args, **kwargs)

        aws_access_key_id = kwargs.get("aws_access_key_id")
        aws_secret_access_key = kwargs.get("aws_secret_access_key")
        region_name = kwargs.get("region_name", "us-east-1")
        cert = kwargs.get('cert', False)
        proxy_url = kwargs.get('proxy_url', None)
        proxies = {"https": proxy_url, "http": proxy_url} if proxy_url else None
        ssl_verify = kwargs.get('ssl_verify', True)
        self.client = Session().client(
            service_name="s3",
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
            region_name=region_name,
            use_ssl=ssl_verify,
            verify=cert,
            config=Config(
                proxies=proxies,
                s3={"addressing_style": "path"} # equivalent to OrdinaryCallingFormat under old boto
            ),
        )


class CustomConnectionV2(BaseConnection):
    def __init__(
        self,
        aws_access_key_id,
        aws_secret_access_key,
        endpoint,
        cert,
        proxy_url
    ):
        self.client = Session().client(
            service_name="s3",
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
            use_ssl=True,
            verify=cert,
            endpoint_url=endpoint,
            config=Config(
                proxies={"https": proxy_url, "http": proxy_url},
            ),
        )

1	from copy import deepcopy	1✔
2	import ssl	1✔
3	import sys	1✔
4	import time	1✔
5
6	from flask import current_app	1✔
7
8	from werkzeug.exceptions import BadRequest	1✔
9	from boto3.session import Session	1✔
10	from botocore.client import Config	1✔
11	from pybossa.cloud_store_api.base_conn import BaseConnection	1✔
12	from os import environ	1✔
13
14
15	def check_store(store):	1✔
16	if not store:	1✔
UNCOV 17	return	×
18
19	store_type = current_app.config.get("S3_CONN_TYPE")	1✔
20	store_type_v2 = current_app.config.get("S3_CONN_TYPE_V2")	1✔
21	if store not in [store_type, store_type_v2]:	1✔
22	raise BadRequest(f"Unsupported store type {store}")	1✔
23
24	def create_connection(**kwargs):	1✔
25	# TODO: remove later
26	v2_access = environ.get("AWS_V2_ACCESS_KEY_ID")	1✔
27	v2_secret = environ.get("AWS_V2_SECRET_ACCESS_KEY")	1✔
28	if v2_access and v2_secret:	1✔
29	masked_v2_secret = f"{v2_secret[:3]}{'x'*(len(v2_secret)-6)}{v2_secret[-3:]}"	×
30	current_app.logger.info("v2_access %s, v2_secret %s", v2_access, masked_v2_secret)	×
31	else:
32	current_app.logger.info("v2_access, v2_secret not found")	1✔
33
34	if kwargs.get("aws_secret_access_key"):	1✔
35	masked_kwargs = {k:v for k, v in kwargs.items()}	1✔
36	secret = kwargs["aws_secret_access_key"]	1✔
37	masked_kwargs["aws_secret_access_key"] = f"{secret[:3]}{'x'*(len(secret)-6)}{secret[-3:]}"	1✔
38	current_app.logger.info(f"create_connection kwargs: %s", str(masked_kwargs))	1✔
39	else:
UNCOV 40	current_app.logger.info(f"create_connection kwargs: %s", str(kwargs))	×
41
42	store = kwargs.pop("store", None)	1✔
43	check_store(store)	1✔
44	store_type_v2 = current_app.config.get("S3_CONN_TYPE_V2")	1✔
45	if store and store == store_type_v2:	1✔
46	current_app.logger.info("Calling CustomConnectionV2")	1✔
47	return CustomConnectionV2(	1✔
48	aws_access_key_id=kwargs.get("aws_access_key_id"),
49	aws_secret_access_key=kwargs.get("aws_secret_access_key"),
50	endpoint=kwargs.get("endpoint"),
51	cert=kwargs.get("cert", False),
52	proxy_url=kwargs.get("proxy_url")
53	)
54
55	current_app.logger.info("Calling CustomConnection")	1✔
56	conn = CustomConnection(**kwargs)	1✔
57	return conn	1✔
58
59
60	class CustomConnection(BaseConnection):	1✔
61
62	def __init__(self, args, *kwargs):	1✔
NEW 63	super().__init__() # super(CustomConnection, self).__init__(args, *kwargs)	×
64
NEW 65	aws_access_key_id = kwargs.get("aws_access_key_id")	×
NEW 66	aws_secret_access_key = kwargs.get("aws_secret_access_key")	×
NEW 67	region_name = kwargs.get("region_name", "us-east-1")	×
NEW 68	cert = kwargs.get('cert', False)	×
NEW 69	proxy_url = kwargs.get('proxy_url', None)	×
NEW 70	proxies = {"https": proxy_url, "http": proxy_url} if proxy_url else None	×
NEW 71	ssl_verify = kwargs.get('ssl_verify', True)	×
NEW 72	self.client = Session().client(	×
73	service_name="s3",
74	aws_access_key_id=aws_access_key_id,
75	aws_secret_access_key=aws_secret_access_key,
76	region_name=region_name,
77	use_ssl=ssl_verify,
78	verify=cert,
79	config=Config(
80	proxies=proxies,
81	s3={"addressing_style": "path"} # equivalent to OrdinaryCallingFormat under old boto
82	),
83	)
84
85
86	class CustomConnectionV2(BaseConnection):	1✔
87	def __init__(	1✔
88	self,
89	aws_access_key_id,
90	aws_secret_access_key,
91	endpoint,
92	cert,
93	proxy_url
94	):
95	self.client = Session().client(	1✔
96	service_name="s3",
97	aws_access_key_id=aws_access_key_id,
98	aws_secret_access_key=aws_secret_access_key,
99	use_ssl=True,
100	verify=cert,
101	endpoint_url=endpoint,
102	config=Config(
103	proxies={"https": proxy_url, "http": proxy_url},
104	),
105	)

bloomberg / pybossa / 19482063855

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