Skip to content

Commit

Permalink
MAINT: Make encryption_key parameter of write_to_stream optional (#1819)
Browse files Browse the repository at this point in the history
This makes PR #1816 smaller

Co-authored-by: exiledkingcc <exiledkingcc@gmail.com>
  • Loading branch information
MartinThoma and exiledkingcc authored Apr 30, 2023
1 parent a6b628a commit 3de03b7
Show file tree
Hide file tree
Showing 9 changed files with 30 additions and 29 deletions.
2 changes: 1 addition & 1 deletion pypdf/_protocols.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ def hash_value(self) -> bytes:
...

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
...

Expand Down
3 changes: 2 additions & 1 deletion pypdf/_writer.py
Original file line number Diff line number Diff line change
Expand Up @@ -1181,6 +1181,7 @@ def _write_pdf_structure(self, stream: StreamType) -> List[int]:
object_positions = []
stream.write(self.pdf_header + b"\n")
stream.write(b"%\xE2\xE3\xCF\xD3\n")

for i, obj in enumerate(self._objects):
obj = self._objects[i]
# If the obj is None we can't write anything
Expand Down Expand Up @@ -1230,7 +1231,7 @@ def _write_trailer(self, stream: StreamType, xref_location: int) -> None:
trailer[NameObject(TK.ID)] = self._ID
if hasattr(self, "_encrypt"):
trailer[NameObject(TK.ENCRYPT)] = self._encrypt
trailer.write_to_stream(stream, None)
trailer.write_to_stream(stream)
stream.write(b_(f"\nstartxref\n{xref_location}\n%%EOF\n")) # eof

def add_metadata(self, infos: Dict[str, Any]) -> None:
Expand Down
20 changes: 10 additions & 10 deletions pypdf/generic/_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ def getObject(self) -> Optional["PdfObject"]: # deprecated
return self.get_object()

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
raise NotImplementedError

Expand All @@ -164,7 +164,7 @@ def clone(
)

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(b"null")

Expand Down Expand Up @@ -218,7 +218,7 @@ def __repr__(self) -> str:
return "True" if self.value else "False"

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
if self.value:
stream.write(b"true")
Expand Down Expand Up @@ -311,7 +311,7 @@ def __ne__(self, other: Any) -> bool:
return not self.__eq__(other)

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(b_(f"{self.idnum} {self.generation} R"))

Expand Down Expand Up @@ -397,7 +397,7 @@ def as_numeric(self) -> float:
return float(self)

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(self.myrepr().encode("utf8"))

Expand Down Expand Up @@ -434,7 +434,7 @@ def as_numeric(self) -> int:
return int(repr(self).encode("utf8"))

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(repr(self).encode("utf8"))

Expand Down Expand Up @@ -488,7 +488,7 @@ def original_bytes(self) -> bytes:
return self

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
bytearr = self
if encryption_key:
Expand Down Expand Up @@ -556,7 +556,7 @@ def get_original_bytes(self) -> bytes:
raise Exception("no information about original bytes")

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
# Try to write the string out as a PDFDocEncoding encoded string. It's
# nicer to look at in the PDF file. Sadly, we take a performance hit
Expand All @@ -570,7 +570,7 @@ def write_to_stream(

bytearr = RC4_encrypt(encryption_key, bytearr)
obj = ByteStringObject(bytearr)
obj.write_to_stream(stream, None)
obj.write_to_stream(stream)
else:
stream.write(b"(")
for c in bytearr:
Expand Down Expand Up @@ -615,7 +615,7 @@ def clone(
)

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(self.renumber()) # b_(renumber(self)))

Expand Down
12 changes: 6 additions & 6 deletions pypdf/generic/_data_structures.py
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ def items(self) -> Iterable[Any]:
return enumerate(self)

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(b"[")
for data in self:
Expand Down Expand Up @@ -338,7 +338,7 @@ def xmpMetadata(self) -> Optional[PdfObject]: # deprecated
return self.xmp_metadata

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(b"<<\n")
for key, value in list(self.items()):
Expand Down Expand Up @@ -783,7 +783,7 @@ def _data(self, value: Any) -> None:
self.__data = value

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
self[NameObject(SA.LENGTH)] = NumberObject(len(self._data))
DictionaryObject.write_to_stream(self, stream, encryption_key)
Expand Down Expand Up @@ -1106,14 +1106,14 @@ def _data(self) -> bytes:
if operator == b"INLINE IMAGE":
new_data.write(b"BI")
dict_text = BytesIO()
operands["settings"].write_to_stream(dict_text, None)
operands["settings"].write_to_stream(dict_text)
new_data.write(dict_text.getvalue()[2:-2])
new_data.write(b"ID ")
new_data.write(operands["data"])
new_data.write(b"EI")
else:
for op in operands:
op.write_to_stream(new_data, None)
op.write_to_stream(new_data)
new_data.write(b" ")
new_data.write(b_(operator))
new_data.write(b"\n")
Expand Down Expand Up @@ -1411,7 +1411,7 @@ def getDestArray(self) -> "ArrayObject": # deprecated
return self.dest_array

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(b"<<\n")
key = NameObject("/D")
Expand Down
2 changes: 1 addition & 1 deletion pypdf/generic/_outline.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

class OutlineItem(Destination):
def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
stream.write(b"<<\n")
for key in [
Expand Down
2 changes: 1 addition & 1 deletion pypdf/xmp.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,7 @@ def rdfRoot(self) -> XmlElement: # deprecated
return self.rdf_root

def write_to_stream(
self, stream: StreamType, encryption_key: Union[None, str, bytes]
self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
) -> None:
self.stream.write_to_stream(stream, encryption_key)

Expand Down
12 changes: 6 additions & 6 deletions tests/test_generic.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ def test_boolean_object(value, expected, tell):
def test_boolean_object_write():
stream = BytesIO()
boolobj = BooleanObject(None)
boolobj.write_to_stream(stream, encryption_key=None)
boolobj.write_to_stream(stream)
stream.seek(0, 0)
assert stream.read() == b"false"

Expand Down Expand Up @@ -213,23 +213,23 @@ def test_name_object(caplog):

# test write
b = BytesIO()
NameObject("/hello").write_to_stream(b, None)
NameObject("/hello").write_to_stream(b)
assert bytes(b.getbuffer()) == b"/hello"

caplog.clear()
b = BytesIO()
NameObject("hello").write_to_stream(b, None)
NameObject("hello").write_to_stream(b)
assert bytes(b.getbuffer()) == b"hello"
assert "Incorrect first char" in caplog.text

caplog.clear()
b = BytesIO()
NameObject("/DIJMAC+Arial Black#1").write_to_stream(b, None)
NameObject("/DIJMAC+Arial Black#1").write_to_stream(b)
assert bytes(b.getbuffer()) == b"/DIJMAC+Arial#20Black#231"
assert caplog.text == ""

b = BytesIO()
NameObject("/你好世界").write_to_stream(b, None)
NameObject("/你好世界").write_to_stream(b)
assert bytes(b.getbuffer()) == b"/#E4#BD#A0#E5#A5#BD#E4#B8#96#E7#95#8C"
assert caplog.text == ""

Expand Down Expand Up @@ -259,7 +259,7 @@ def test_destination_fit_v():
def test_outline_item_write_to_stream():
stream = BytesIO()
oi = OutlineItem(NameObject("title"), NullObject(), Fit.fit_vertically(left=0))
oi.write_to_stream(stream, None)
oi.write_to_stream(stream)
stream.seek(0, 0)
assert stream.read() == b"<<\n/Title (title)\n/Dest [ null /FitV 0.0 ]\n>>"

Expand Down
2 changes: 1 addition & 1 deletion tests/test_protocols.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,4 +12,4 @@ def test_pdfobjectprotocol():
assert o._reference_clone(None, None) is None
assert o.get_object() is None
assert o.hash_value() is None
assert o.write_to_stream(None, None) is None
assert o.write_to_stream(None) is None
4 changes: 2 additions & 2 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -904,14 +904,14 @@ def test_extra_test_iss1541():
cs = ContentStream(reader.pages[0]["/Contents"], None, None)
cs.operations.insert(-1, ([], b"EMC"))
bu = BytesIO()
cs.write_to_stream(bu, None)
cs.write_to_stream(bu)
bu.seek(0)
ContentStream(read_object(bu, None, None), None, None).operations

cs = ContentStream(reader.pages[0]["/Contents"], None, None)
cs.operations.insert(-1, ([], b"E!C"))
bu = BytesIO()
cs.write_to_stream(bu, None)
cs.write_to_stream(bu)
bu.seek(0)
with pytest.raises(PdfReadError) as exc:
ContentStream(read_object(bu, None, None), None, None).operations
Expand Down

0 comments on commit 3de03b7

Please sign in to comment.