[Git][debian-gis-team/pyshp][upstream] New upstream version 3.1.4
Bas Couwenberg (@sebastic)
gitlab at salsa.debian.org
Tue Jun 30 05:03:18 BST 2026
Bas Couwenberg pushed to branch upstream at Debian GIS Project / pyshp
Commits:
9690919b by Bas Couwenberg at 2026-06-30T05:56:09+02:00
New upstream version 3.1.4
- - - - -
5 changed files:
- README.md
- changelog.txt
- pyproject.toml
- src/shapefile.py
- tests/hypothesis_tests.py
Changes:
=====================================
README.md
=====================================
@@ -8,8 +8,8 @@ The Python Shapefile Library (PyShp) reads and writes ESRI Shapefiles in pure Py
- **Author**: [Joel Lawhead](https://github.com/GeospatialPython)
- **Maintainers**: [James Parrott](https://github.com/JamesParrott) & [Karim Bahgat](https://github.com/karimbahgat)
-- **Version**: 3.1.3
-- **Date**: 25th June 2026
+- **Version**: 3.1.4
+- **Date**: 29th June 2026
- **License**: [MIT](https://github.com/GeospatialPython/pyshp/blob/master/LICENSE.TXT)
## Contents
@@ -93,6 +93,14 @@ part of your geospatial project.
# Version Changes
+## 3.1.4
+### Bug fix
+ - Fix bug causing dates supplied as length 8 strings of digits to be encoded by the custom encoding, not ascii.
+
+### Testing
+ - Test other codecs (ascii and UTF-8, UTF-16 & UTF-32 so far).
+ - Test all available codecs in CI (92 of them).
+
## 3.1.3
- Restore faster text writing paths for single-byte Ascii encodings, and Utf-8.
@@ -522,7 +530,7 @@ Finally, you can use all of the above methods to read shapefiles directly from t
>>> # from a zipped shapefile on website
- >>> sf = shapefile.Reader("https://geodata.ucdavis.edu/gadm/gadm4.1/shp/gadm41_ATA_shp.zip")
+ >>> sf = shapefile.Reader("https://github.com/JamesParrott/PyShp_test_shapefile/raw/refs/heads/main/gadm41_ATA_shp.zip")
>>> # from a shapefile collection of files in a github repository
>>> sf = shapefile.Reader("https://pubs.usgs.gov/of/2000/of00-006/gisdata/coverage/airports.shp")
=====================================
changelog.txt
=====================================
@@ -1,3 +1,11 @@
+
+VERSION 3.1.4
+2026-06-29
+ * Fix bug causing dates supplied as length 8 strings to be encoded by the custom codec, not ascii
+2026-06-27
+ * Test other codecs (ascii and UTF-8, UTF-16 & UTF-32 so far).
+ * Test all other codecs in CI
+
VERSION 3.1.3
2026-06-25
=====================================
pyproject.toml
=====================================
@@ -74,6 +74,7 @@ markers = [
"network: marks tests requiring network access",
"slow: marks other tests that cause bottlenecks",
"hypothesis: tests that require hypothesis",
+ "hypothesis_dbf: hypothesis tests that test dbf functionality",
]
python_files = "test_*.py *_test.py *_tests.py"
=====================================
src/shapefile.py
=====================================
@@ -8,7 +8,7 @@ Compatible with Python versions >=3.9
from __future__ import annotations
-__version__ = "3.1.3"
+__version__ = "3.1.4"
import abc
import array
@@ -22,13 +22,13 @@ import tempfile
import time
import warnings
import zipfile
-from collections.abc import Container, Iterable, Iterator, Mapping, Reversible, Sequence
+from collections.abc import Container, Iterable, Iterator, Reversible, Sequence
from contextlib import AbstractContextManager, ExitStack
from datetime import date, datetime
from os import PathLike
from pathlib import Path
from struct import Struct, calcsize, error, pack, unpack
-from types import MappingProxyType, TracebackType
+from types import TracebackType
from typing import (
IO,
Any,
@@ -283,34 +283,59 @@ def _truncate_utf8_str(
)
+ at functools.cache
def _BOM_and_dbf_decoded_pad_bytes(
+ pad_byte: Literal[b" ", b"\x00"],
encoding: str = "utf8",
-) -> tuple[bytes, Mapping[str, bytes]]:
+) -> tuple[bytes, dict[str, bytes], dict[str, bytes], dict[str, bytes]]:
try:
BOM = "".encode(encoding)
except UnicodeEncodeError:
BOM = b""
- tuples: list[tuple[str, bytes]] = []
- for pad_byte_str, N in {b" ": 5, b"\x00": 5, b" \x00": 2}.items():
- # Max code unit size under UTF-8, UTF-16, and UTF-32 is 4 bytes.
- for n in range(1, N):
- pad_bytes = pad_byte_str * n
+ N: int # code-unit size in bytes (possible length of
+ # byte strings, that a single code point could encode to)
+ if encoding.lower().startswith("utf32"):
+ N = 4
+ elif encoding.lower().startswith("utf16"):
+ # Null bytes and ascii spaces don't encode to Surrogate-pairs
+ N = 2
+ else:
+ # Both Ascii and UTF-8 handled here (UTF-8 is backward compatible with ascii)
+ N = 1
+
+ def decoded_code_points_and_bytes(
+ pad_byte_strs: Iterable[bytes],
+ ) -> dict[str, bytes]:
+ retval = {}
+ for pad_bytes in pad_byte_strs:
try:
s: str = (BOM + pad_bytes).decode(encoding)
except UnicodeDecodeError:
continue
- tuples.append((s, pad_bytes))
- break
- return BOM, MappingProxyType(dict(tuples))
+ retval[s] = pad_bytes
+ return retval
+
+ # Max code unit size under UTF-8, UTF-16, and UTF-32 is 4 bytes.
+ if pad_byte == b"\x00":
+ # Just checking the field, in which asii spaces are technically valid
+ # even though PyShp historically has converted them to underscores
+ return BOM, {}, decoded_code_points_and_bytes([b"\x00" * N]), {}
+ else:
+ pad_byte_strs = [b" " * i + b"\x00" * (N - i) for i in range(N + 1)]
+
+ all_ascii_spaces = decoded_code_points_and_bytes([b" "])
+ mixed = decoded_code_points_and_bytes(pad_byte_strs)
+ all_null_bytes = decoded_code_points_and_bytes([b"\x00"])
+
+ return BOM, all_ascii_spaces, mixed, all_null_bytes
def _encode_dbf_string(
s: str,
size: int,
- decode: Decoder | None,
- pad_byte: bytes,
- decoded_pad_bytes: Mapping[str, bytes],
+ pad_byte: Literal[b" ", b"\x00"],
+ decode: Decoder | None = None,
encoding: str = "utf8",
encodingErrors: str = "strict",
strict: bool = True,
@@ -337,7 +362,7 @@ def _encode_dbf_string(
if len(encoded) <= size:
if i <= N - 1:
msg = (
- f"Dropped {N - i} code points (e.g. characters)! "
+ f"Dropped {N - i} out of {N} code points (e.g. characters)! "
f"{s} was truncated to {trimmed} (discarding: {s[i:]}), "
f"in order to encode it under {size} bytes for the field or field name. "
f"Used: {encoding=} and {encodingErrors=}. "
@@ -358,19 +383,43 @@ def _encode_dbf_string(
f"to a short enough byte string, using {encoding=}, {encodingErrors=} ({BOM=!r})"
)
- for suffix, pad_bytes in decoded_pad_bytes.items():
- if s.endswith(suffix):
- msg = (
- f"Under the given encoding: {encoding}, "
- f" the text (field name or 'C' or 'M' field): {s!r} "
- f" ends with {suffix!r}, which "
- f"encodes to the pad bytes: {pad_bytes!r}. "
- "The real end of the actual data may be earlier. "
- )
- if strict:
- raise DbfStringDataLoss(msg)
- warnings.warn(msg, category=PossibleDataLoss)
- break
+ _BOM, all_first, mixed, all_last = _BOM_and_dbf_decoded_pad_bytes(
+ pad_byte, encoding
+ )
+ already_warned = False
+
+ def check_and_trim(decoded_pad_bytes: dict[str, bytes]) -> None:
+
+ nonlocal trimmed, already_warned
+
+ for suffix, pad_bytes in decoded_pad_bytes.items():
+ if not suffix:
+ continue
+ if len(suffix) >= 2:
+ raise ValueError(
+ f"Multiple code points: {suffix} encoded to: {pad_bytes!r} under {encoding=}"
+ )
+ if trimmed.endswith(suffix):
+ msg = (
+ f"Under the given encoding: {encoding}, after truncation to {size} bytes,"
+ f" the remaining text (field name or 'C' or 'M' field): {trimmed!r} "
+ f" ends with {suffix!r}, which "
+ f"encodes to the pad bytes: {pad_bytes!r}. "
+ "The real end of the actual data may be earlier. "
+ )
+ if strict:
+ raise DbfStringDataLoss(msg)
+ if not already_warned:
+ warnings.warn(msg, category=PossibleDataLoss)
+ already_warned = True
+ if len(set(pad_bytes)) == 1: # all same byte => strip all code points
+ trimmed = trimmed.rstrip(suffix)
+ else:
+ trimmed = trimmed.removesuffix(suffix)
+
+ check_and_trim(all_last)
+ check_and_trim(mixed)
+ check_and_trim(all_first)
if len(encoded) < size:
padded = encoded.ljust(size, pad_byte)
@@ -382,7 +431,8 @@ def _encode_dbf_string(
with warnings.catch_warnings():
warnings.simplefilter("ignore")
- # TODO: Fuzz test this to see what it actually catches.
+ # TODO: Fuzz test this to see what it actually catches,
+ # as it makes encoding much slower.
decoded = decode(
b=padded,
encoding=encoding,
@@ -405,7 +455,7 @@ def _encode_dbf_string(
def _try_to_decode_dbf_name_or_text_field(
b: bytes,
- pad_bytes: bytes, # Pad bytes will be trimmed (from the R of b) in their order in the byte-string
+ pad_bytes: bytes, # Pad bytes will be trimmed from the RHS (end) of b.
encoding: str = "utf8",
encodingErrors: str = "strict",
) -> str:
@@ -523,7 +573,6 @@ class Field(NamedTuple):
cls,
name: str,
*,
- decoded_pad_bytes: Mapping[str, bytes],
field_type: str | bytes | FieldTypeT = "C",
size: int = 50,
decimal: int = 0,
@@ -532,14 +581,16 @@ class Field(NamedTuple):
strict: bool = False,
) -> Field:
+ name = str(name)
+
if "\x00" in name:
msg = (
- "Field names should not contain null characters "
+ "Field names ought not contain null characters, "
"as null bytes are used for padding in the header. "
f"Got: {name=} "
)
if strict:
- raise dbfFileException(msg)
+ raise DbfStringDataLoss(msg)
warnings.warn(msg, category=PossibleDataLoss)
try:
@@ -567,11 +618,10 @@ class Field(NamedTuple):
# Only use the portion of the name that we are able to encode to
# 10 bytes or less.
_encoded_name, trimmed_name = cls.trim_name_until_encodable(
- name=str(name),
+ name=name,
encoding=encoding,
encodingErrors=encodingErrors,
strict=strict,
- decoded_pad_bytes=decoded_pad_bytes,
)
# A doctest in README.md previously passed in a string ('40') for size,
@@ -582,29 +632,28 @@ class Field(NamedTuple):
# Raise Exception or trigger warning early, before user adds more fields
# (fields are only written when first record added, and on close)
+ # Tests field_type, size and decimal. Name already tested and cached above.
inst.encode_field_descriptor(
encoding=encoding,
encodingErrors=encodingErrors,
strict=strict,
- decoded_pad_bytes=decoded_pad_bytes,
)
return inst
@classmethod
+ @functools.cache
def trim_name_until_encodable(
cls,
name: str,
encoding: str = "utf8",
encodingErrors: str = "strict",
strict: bool = False,
- decoded_pad_bytes: Mapping[str, bytes] = {},
) -> tuple[bytes, str]:
return _encode_dbf_string(
s=name,
size=10,
decode=cls.decode_name,
pad_byte=b"\x00",
- decoded_pad_bytes=decoded_pad_bytes,
encoding=encoding,
encodingErrors=encodingErrors,
strict=strict,
@@ -615,7 +664,6 @@ class Field(NamedTuple):
encoding: str = "utf8",
encodingErrors: str = "strict",
strict: bool = False,
- decoded_pad_bytes: Mapping[str, bytes] = {},
) -> bytes:
# encoded_name = self.name.encode(encoding, encodingErrors)
# encoded_name = encoded_name[:10].ljust(10, b"\x00")
@@ -624,7 +672,6 @@ class Field(NamedTuple):
encoding=encoding,
encodingErrors=encodingErrors,
strict=strict,
- decoded_pad_bytes=decoded_pad_bytes,
)
encoded_field_type = self.field_type.encode("ascii")
@@ -4219,7 +4266,12 @@ class DbfWriter(_HasCheckedWriteableFile):
self.recNum = 0
self._is_utf8 = encoding.replace("-", "").replace("_", "").lower() == "utf8"
- self._BOM, self._decoded_pad_bytes = _BOM_and_dbf_decoded_pad_bytes(encoding)
+ (
+ self._BOM,
+ self._decoded_ascii_spaces,
+ self._decoded_mixed_bytes,
+ self._decoded_null_bytes,
+ ) = _BOM_and_dbf_decoded_pad_bytes(b" ", encoding)
def field(
# Types of args should match *Field
@@ -4242,7 +4294,6 @@ class DbfWriter(_HasCheckedWriteableFile):
encoding=self.encoding,
encodingErrors=self.encodingErrors,
strict=self.strict,
- decoded_pad_bytes=self._decoded_pad_bytes,
)
self.fields.append(field)
@@ -4287,7 +4338,6 @@ class DbfWriter(_HasCheckedWriteableFile):
encoding=self.encoding,
encodingErrors=self.encodingErrors,
strict=self.strict,
- decoded_pad_bytes=self._decoded_pad_bytes,
)
)
@@ -4386,7 +4436,7 @@ class DbfWriter(_HasCheckedWriteableFile):
elif value in MISSING:
str_val = "0" * 8 # QGIS NULL for date type
elif isinstance(value, str) and len(value) == 8:
- pass # value is already a date string
+ str_val = value
else:
raise ShapefileException(
f"Could not read as date: {value}. "
@@ -4439,14 +4489,21 @@ class DbfWriter(_HasCheckedWriteableFile):
)
if self.strict:
raise DbfStringDataLoss(msg)
- warnings.warn(msg)
+ warnings.warn(msg, category=PossibleDataLoss)
+
+ depadded = trimmed
+ for byte in [b"\x00", b" "]:
+ try:
+ decoded_pad_byte = byte.decode(
+ self.encoding, self.encodingErrors
+ )
+ except UnicodeDecodeError:
+ continue
+ depadded = depadded.rstrip(decoded_pad_byte)
- # TODO: Handle decoded_pad_bytes longer than 1
- pad_bytes = "".join(self._decoded_pad_bytes)
- depadded = trimmed.rstrip(pad_bytes)
if len(depadded) < len(trimmed):
msg = (
- f"Trimmed: {trimmed}, stringified: {str_val} of data: {value} "
+ f"Trimmed: {trimmed!r}, stringified: {str_val!r} of data: {value!r} "
f"ends in decoded pad bytes or decoded null bytes. "
"Data encoded as null bytes and pad bytes will probably not "
"be recovered by applications reading the Shapefile or dbf file "
@@ -4454,7 +4511,7 @@ class DbfWriter(_HasCheckedWriteableFile):
)
if self.strict:
raise DbfStringDataLoss(msg)
- warnings.warn(msg)
+ warnings.warn(msg, category=PossibleDataLoss)
encoded = encoded.ljust(size)
else:
@@ -4463,7 +4520,6 @@ class DbfWriter(_HasCheckedWriteableFile):
size=size,
decode=_decode_C_or_M_field if self.strict else None,
pad_byte=b" ",
- decoded_pad_bytes=self._decoded_pad_bytes,
encoding=self.encoding,
encodingErrors=self.encodingErrors,
strict=self.strict,
@@ -4886,6 +4942,14 @@ class Writer(_HasExitStack):
def fields(self, value: list[Field]) -> None:
self.dbf_writer.fields = value
+ @property
+ def strict(self) -> bool:
+ return self.dbf_writer.strict
+
+ @strict.setter
+ def strict(self, value: bool) -> None:
+ self.dbf_writer.strict = value
+
@property
def recNum(self) -> int:
if not self._dbf_writer:
=====================================
tests/hypothesis_tests.py
=====================================
@@ -1,9 +1,12 @@
from __future__ import annotations
+import contextlib
import datetime
import io
import itertools
+import os
import string
+import warnings
import pytest
from hypothesis import HealthCheck, given, settings, reproduce_failure
@@ -25,6 +28,18 @@ from hypothesis.strategies import (
import shapefile as shp
+IN_CI = bool(os.getenv("CI") or os.getenv("GITHUB_ACTIONS"))
+
+ at contextlib.contextmanager
+def ignore_warnings(category=None):
+ with warnings.catch_warnings():
+ if category:
+ warnings.simplefilter("ignore", category)
+ else:
+ warnings.simplefilter("ignore")
+ yield
+
+
float_nums = floats(allow_nan=False, allow_infinity=False)
xs = float_nums
ys = float_nums
@@ -441,7 +456,7 @@ shape_codes_names_and_strategies = [
(31, "MultiPatch", multipatch),
]
-def code_and_shape_strat_from_triple(t):
+def code_and_shape_strategy_from_triple(t):
x, _name, shapes = t
return tuples(
just(x),
@@ -451,12 +466,40 @@ def code_and_shape_strat_from_triple(t):
max_size=MAX_NUM_SHAPES,
),
)
-codes_and_shapes_strats = [
- code_and_shape_strat_from_triple(t)
+codes_and_shapes_strategies = [
+ code_and_shape_strategy_from_triple(t)
for t in shape_codes_names_and_strategies
]
-codes_and_shapes = one_of(codes_and_shapes_strats)
+codes_and_shapes = one_of(codes_and_shapes_strategies)
+
+
+def _assert_reader_matches_expected_shapes(r, code_ex, expected_shapes):
+ assert r.shapeType == code_ex
+
+ for actual, expected in itertools.zip_longest(r.shapes(), expected_shapes):
+
+ assert isinstance(actual, (shp.SHAPE_CLASS_FROM_SHAPETYPE[code_ex], shp.NullShape))
+ assert actual.points_3D == expected.points_3D
+ # Don't assert actual.oid == expected.oid it's defined by
+ # actual.oid indicates the order actual was written in, expected.oid
+ # is not currently encoded (as we'd have to resort the entire Shapefile after each shape)
+ assert actual.parts == expected.parts, f"{type(actual.parts)=}, {type(expected.parts)=}"
+
+ if (m := getattr(actual, "m", None)):
+ assert m == expected.m, f"{type(m)=}, {type(expected.m)=}"
+ else:
+ assert not hasattr(expected, "m")
+
+ if (z := getattr(actual, "z", None)):
+ assert z == expected.z, f"{type(z)=}, {type(expected.z)=}"
+ else:
+ assert not hasattr(expected, "z")
+
+ if (partTypes := getattr(actual, "partTypes", None)):
+ assert actual.partTypes == expected.partTypes, f"{type(actual.partTypes)=}, {type(expected.partTypes)=}"
+ else:
+ assert not hasattr(expected, "partTypes")
@pytest.mark.hypothesis
@given(codes_and_shapes=codes_and_shapes)
@@ -470,31 +513,8 @@ def test_shp_reader_writer_roundtrip(codes_and_shapes)-> None:
w.shape(shape)
with shp.ShpReader(shp=stream) as r:
- assert r.shapeType == code_ex
-
- for actual, expected in itertools.zip_longest(r.shapes(), expected_shapes):
-
- assert isinstance(actual, (shp.SHAPE_CLASS_FROM_SHAPETYPE[code_ex], shp.NullShape))
- assert actual.points_3D == expected.points_3D
- # Don't assert actual.oid == expected.oid it's defined by
- # actual.oid indicates the order actual was written in, expected.oid
- # is not currently encoded (as we'd have to resort the entire Shapefile after each shape)
- assert actual.parts == expected.parts, f"{type(actual.parts)=}, {type(expected.parts)=}"
+ _assert_reader_matches_expected_shapes(r, code_ex, expected_shapes)
- if (m := getattr(actual, "m", None)):
- assert m == expected.m, f"{type(m)=}, {type(expected.m)=}"
- else:
- assert not hasattr(expected, "m")
-
- if (z := getattr(actual, "z", None)):
- assert z == expected.z, f"{type(z)=}, {type(expected.z)=}"
- else:
- assert not hasattr(expected, "z")
-
- if (partTypes := getattr(actual, "partTypes", None)):
- assert actual.partTypes == expected.partTypes, f"{type(actual.partTypes)=}, {type(expected.partTypes)=}"
- else:
- assert not hasattr(expected, "partTypes")
@pytest.mark.hypothesis
@@ -533,15 +553,53 @@ DBF_FIELD_TYPES = {
"D": {"min_length": 8, "max_length": 8},
}
+
+ENCODINGS = [
+ "ascii",
+ "latin1",
+ "utf-8",
+ "utf-16-be",
+ "utf-16-le",
+ "utf-32-le",
+ "cp1140",
+]
+
+def _encodings() -> set[str]:
+ from encodings.aliases import aliases
+ encs = set()
+ for enc in aliases.values():
+ if enc in encs:
+ continue
+ try:
+ "".encode(enc)
+ except (UnicodeEncodeError, LookupError):
+ continue
+ encs.add(enc)
+ return encs
+# assert _encodings() == {'utf_16_le', 'iso8859_7', 'cp437', 'iso2022_jp_3', 'shift_jis', 'cp775', 'cp1140',
+# 'cp861', 'iso8859_11', 'iso8859_9', 'euc_jp', 'utf_16', 'cp950', 'mac_cyrillic', 'mac_turkish', 'iso2022_jp_1', 'iso8859_10',
+# 'iso2022_jp_2004', 'cp866', 'mac_greek', 'hz', 'cp1257', 'cp037', 'cp863', 'iso8859_4', 'utf_16_be', 'gb18030', 'cp1250',
+# 'cp850', 'iso8859_5', 'shift_jisx0213', 'iso8859_8', 'cp273', 'euc_jisx0213', 'cp932', 'cp862', 'tis_620', 'cp1125', 'koi8_r',
+# 'cp874', 'cp1026', 'cp1252', 'cp858', 'cp865', 'gb2312', 'iso8859_15', 'cp857', 'cp860', 'iso2022_jp', 'iso2022_jp_ext',
+# 'ascii', 'cp1254', 'cp424', 'cp855', 'hp_roman8', 'mac_latin2', 'euc_jis_2004', 'euc_kr', 'cp1256', 'shift_jis_2004',
+# 'utf_32_le', 'gbk', 'cp869', 'iso8859_13', 'iso8859_3', 'big5', 'cp1258', 'cp1253', 'latin_1', 'cp864', 'utf_8',
+# 'iso2022_kr', 'cp1251', 'cp1255', 'mac_iceland', 'kz1048', 'iso8859_14', 'utf_32_be', 'ptcp154', 'iso8859_6', 'mac_roman',
+# 'utf_32', 'iso2022_jp_2', 'iso8859_16', 'mbcs', 'cp500', 'iso8859_2', 'cp949', 'cp852', 'utf_7', 'big5hkscs', 'johab'}
+
+encodings = sampled_from(list(_encodings())) # if IN_CI else ENCODINGS)
+
+
@composite
-def dbf_fields(draw):
+def _dbf_fields_strategy(draw, encoding: str) -> dict[str, str | int]:
field_type, bounds_dict = draw(sampled_from(list(DBF_FIELD_TYPES.items())))
name = draw(
text(
alphabet=characters(
- codec="ascii",
- exclude_categories=["Z", "C"] # Z - Whitespace, C - Control chars++
+ codec=encoding,
+ # https://en.wikipedia.org/wiki/Unicode_character_property#General_Category
+ exclude_categories=["Cs", "Co", "Cn"], # Cs - surrogates
+ # exclude_characters=[" "],
),
min_size=1,
max_size=10,
@@ -550,27 +608,79 @@ def dbf_fields(draw):
max_length = bounds_dict.get("max_length", 254)
min_length = bounds_dict.get("min_length", 1)
+ if field_type in {"C", "M"}:
+ # Make sure field is big enough to store any BOM
+ # used by non-endianness specified codecs
+ # (e.g. utf-16 and utf-32)
+ min_length = max(min_length, len("".encode(encoding)))
+ max_length = max(max_length, min_length)
max_decimal = bounds_dict.get("max_decimal", 0)
size = draw(integers(min_value=min_length, max_value=max_length))
decimal = draw(integers(min_value=0, max_value=max(0,min(size - 3, max_decimal))))
-
return {"name": name, "field_type": field_type, "size": size, "decimal": decimal}
+ at composite
+def encodings_and_dbf_fields(draw):
+ encoding = draw(encodings)
+ fields_strategy = _dbf_fields_strategy(encoding)
+ field = draw(fields_strategy)
+ return encoding, field
+
+def _get_fields_w_context(fields, codec, strict=False):
+ for field in fields:
+ if (len(field["name"].encode(codec)) > 10 or
+ "\x00" in field["name"] or
+ (" " in field["name"] and not strict)
+ ):
+ if strict:
+ return pytest.raises(shp.DbfStringDataLoss), True
+ return pytest.warns(shp.PossibleDataLoss), False
+ return contextlib.nullcontext(), False
+
+def _get_fields_r_context(codec):
+ # In utf-16-le and utf-32-le, many low code points encode
+ # to code units ending in null bytes, causing warnings in field
+ # names (which use trailing null bytes for padding).
+ normalised = codec.lower().replace("-","").replace("_","")
+ if (any(normalised.startswith(prefix) for prefix in ["utf16", "utf32"]) and
+ not codec.lower().endswith("-be")):
+
+ return ignore_warnings(shp.PossibleDataLoss)
+ return contextlib.nullcontext()
+
+
@pytest.mark.hypothesis
- at given(field_kwargs=dbf_fields())
-def test_dbf_Field_roundtrips(
- field_kwargs: dict,
-) -> None:
- BOM, decoded_pad_bytes = shp._BOM_and_dbf_decoded_pad_bytes()
+ at pytest.mark.hypothesis_dbf
+ at settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
+ at given(encoding_and_dbf_field=encodings_and_dbf_fields())
+def test_dbf_Field_roundtrips(encoding_and_dbf_field: dict) -> None:
- expected = shp.Field.from_unchecked(decoded_pad_bytes=decoded_pad_bytes,**field_kwargs)
+ encoding, field_kwargs = encoding_and_dbf_field
+
+ w_context, error_expected = _get_fields_w_context([field_kwargs], encoding, strict=True)
+
+ with w_context:
+ expected = shp.Field.from_unchecked(
+ encoding=encoding,
+ strict=True,
+ **field_kwargs,
+ )
+ encoded = expected.encode_field_descriptor(encoding=encoding, strict=True)
+ if error_expected:
+ return
stream = io.BytesIO()
- encoded = expected.encode_field_descriptor(strict=True)
stream.write(encoded)
stream.seek(0)
- actual = shp.Field.from_byte_stream(stream)
+
+
+ with _get_fields_r_context(encoding):
+ actual = shp.Field.from_byte_stream(
+ stream,
+ encoding=encoding,
+ )
+
assert isinstance(actual, shp.Field)
assert actual.name == expected.name
assert actual[1:] == expected[1:]
@@ -578,7 +688,7 @@ def test_dbf_Field_roundtrips(
ascii_printable = string.ascii_letters + string.digits + string.punctuation + " "
-def record_value_for_field(name: str, field_type: str, size: int, decimal: int = 0):
+def record_value_for_field(name: str, field_type: str, size: int, decimal: int, encoding: str):
if field_type == "C":
return text(
@@ -610,67 +720,165 @@ def record_value_for_field(name: str, field_type: str, size: int, decimal: int =
raise ValueError(f"Unsupported: {field_type=}")
-def _dbf_fields_and_record_strategy(
+def _dbf_encoding_fields_and_record_strategy(
draw,
- max_fields=10, # In DbfWriter.__init__, max_num_fields: int = 2046,
- max_records=20,
+ max_fields: int=10, # In DbfWriter.__init__, max_num_fields: int = 2046,
):
- fields = draw(lists(dbf_fields(), min_size=1, max_size=max_fields))
+ encoding = draw(encodings)
- record_strategy = tuples(*(record_value_for_field(**field) for field in fields))
+ fields = draw(lists(_dbf_fields_strategy(encoding), min_size=1, max_size=max_fields))
- return fields, record_strategy
+ record_strategy = tuples(*(record_value_for_field(encoding=encoding, **field) for field in fields))
+
+ return encoding, fields, record_strategy
@composite
-def dbf_fields_and_records(
+def dbf_encoding_fields_and_records(
draw,
max_fields=10, # In DbfWriter.__init__, max_num_fields: int = 2046,
max_records=20,
):
- fields, record_strategy = _dbf_fields_and_record_strategy(draw, max_fields, max_records)
+ encoding, fields, record_strategy = _dbf_encoding_fields_and_record_strategy(draw, max_fields)
records = draw(lists(record_strategy, min_size=0, max_size=max_records))
- return fields, records
+ return encoding, fields, records
+
+
+def _assert_reader_matches_expected_fields(r, expected_fields, writer_strict):
+ assert len(expected_fields) == len(r.data_fields), f"{expected_fields=}, {r.data_fields=}"
+
+ for f_r, f_w in zip(r.data_fields, expected_fields):
+ expected_name = f_w["name"]
+ if not writer_strict:
+ expected_name = expected_name.replace(" ", "_")
+ expected_name = expected_name.rstrip("\x00")
+ assert expected_name.startswith(f_r.name), f"{expected_name=}, {f_r.name=}"
+ actual_field_dict = f_r._asdict()
+ for k in ("field_type", "size", "decimal"):
+ assert actual_field_dict[k] == f_w[k], f"{k=}, {actual_field_dict[k]=}, {f_w[k]=}"
+
+def _assert_reader_matches_expected_records(r, fields, written_records):
+ actual_records = r.records()
+ expected_records = [rec for rec in written_records if rec is not None]
+ assert len(expected_records) == len(actual_records), f"{expected_records=}, {actual_records=}"
+ for exp_rec, actual_rec in zip(expected_records, actual_records):
+ for expected, actual, field in itertools.zip_longest(exp_rec, actual_rec, fields):
+ field_type = field["field_type"]
+ decimal = field["decimal"]
+ if field_type == "D":
+ if isinstance(expected, datetime.date):
+ expected = expected.strftime("%Y%m%d")
+ if isinstance(actual, datetime.date):
+ actual = actual.strftime("%Y%m%d")
+ elif field_type in ("N", "F") and decimal >= 1:
+ expected = float(format(expected, f".{decimal}f"))
+ assert actual == expected, f"{actual=}, {expected=}, {field_type=}, {type(actual)=}, {type(expected)=}"
+
+
+def _write_fields_and_records_to_strict(w, fields, records):
+
+ field_indices, written_records = set(), []
+
+
+ for i, field in enumerate(fields):
+ try:
+ w.field(**field)
+ except shp.DbfStringDataLoss:
+ pass
+ else:
+ field_indices.add(i)
+
+ if not field_indices:
+ return None, None
+
+
+ for record in records:
+ rec_list = [
+ val
+ for i, val in enumerate(record)
+ if i in field_indices
+ ]
+ try:
+ w.record(*rec_list)
+ except shp.DbfStringDataLoss:
+ written_records.append(None)
+ else:
+ written_records.append(rec_list)
+ written_fields = [field for i, field in enumerate(fields) if i in field_indices]
+
+ return written_fields, written_records
+
@pytest.mark.hypothesis
- at given(fields_and_records=dbf_fields_and_records())
-def test_dbf_reader_writer_roundtrip(fields_and_records)-> None:
- fields, records = fields_and_records
+ at pytest.mark.hypothesis_dbf
+ at settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
+ at given(codec_fields_and_records=dbf_encoding_fields_and_records())
+def test_dbf_reader_writer_roundtrip(codec_fields_and_records)-> None:
+ codec, fields, records = codec_fields_and_records
stream = io.BytesIO()
- written_records = []
- with shp.DbfWriter(dbf=stream, strict=True) as dbf_w:
- for field in fields:
- dbf_w.field(**field)
- for record in records:
- try:
- dbf_w.record(*record)
- except shp.DbfStringDataLoss:
- pass
- else:
- written_records.append(record)
-
-
- with shp.DbfReader(dbf=stream) as r:
- actual_fields = iter(r.fields)
- next(actual_fields) # skip deletion flag
- for f_r, f_w in itertools.zip_longest(actual_fields, fields):
- actual_field_dict = f_r._asdict()
- for k in ("field_type", "size", "decimal"):
- assert actual_field_dict[k] == f_w[k], f"{k=}, {actual_field_dict[k]=}, {f_w[k]=}"
- for exp_rec, actual_rec in itertools.zip_longest(written_records, r.records()):
- for expected, actual, field in itertools.zip_longest(exp_rec, actual_rec, fields):
- field_type = field["field_type"]
- decimal = field["decimal"]
- if field_type == "D":
- if isinstance(expected, datetime.date):
- expected = expected.strftime("%Y%m%d")
- if isinstance(actual, datetime.date):
- actual = actual.strftime("%Y%m%d")
- elif field_type in ("N", "F") and decimal >= 1:
- expected = float(format(expected, f".{decimal}f"))
- assert actual == expected, f"{actual=}, {expected=}, {field_type=}, {type(actual)=}, {type(expected)=}"
+
+ # pytest.raises and pytest.warns can obscure other
+ # exceptions inside them, when iterating on the test code
+ w = shp.DbfWriter(dbf=stream, encoding=codec, strict=True)
+
+ written_fields, written_records = _write_fields_and_records_to_strict(w, fields, records)
+
+ if not written_fields or written_records is None:
+ return
+
+ w.close()
+
+
+ with _get_fields_r_context(codec), shp.DbfReader(dbf=stream, encoding=codec) as r:
+ _assert_reader_matches_expected_fields(r, written_fields, True)
+ _assert_reader_matches_expected_records(r, written_fields, written_records)
+
+
+ at composite
+def codes_codecs_fields_shapes_and_records(draw):
+ code, shapes = draw(codes_and_shapes)
+ encoding, fields, records_strategy = _dbf_encoding_fields_and_record_strategy(draw, max_fields=10)
+ N = len(shapes)
+ records = [draw(records_strategy) for _ in range(N)]
+
+ return code, encoding, fields, shapes, records
+
+
+ at pytest.mark.hypothesis
+ at pytest.mark.hypothesis_dbf
+ at settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
+ at given(codes_codecs_fields_shapes_and_records=codes_codecs_fields_shapes_and_records())
+def test_shapefile_reader_writer_roundtrip(codes_codecs_fields_shapes_and_records)-> None:
+
+ code_ex, encoding, fields, shapes, records = codes_codecs_fields_shapes_and_records
+ streams = {"shp" : io.BytesIO(), "shx" : io.BytesIO(), "dbf" : io.BytesIO(),}
+ w = shp.Writer(shapeType = code_ex, encoding=encoding, strict=True, **streams)
+
+ expected_shapes = []
+
+ written_fields, written_records = _write_fields_and_records_to_strict(w, fields, records)
+
+ if not written_fields:
+ try:
+ w.close()
+ except shp.dbfFileException:
+ pass
+ return
+
+ for shape, written_record in zip(shapes, written_records):
+ if written_record is None:
+ continue
+ w.shape(shape)
+ expected_shapes.append(shape)
+
+ w.close()
+
+ with _get_fields_r_context(encoding), shp.Reader(encoding=encoding, **streams) as r:
+ _assert_reader_matches_expected_fields(r, written_fields, True)
+ _assert_reader_matches_expected_records(r, written_fields, written_records)
+ _assert_reader_matches_expected_shapes(r, code_ex, expected_shapes)
View it on GitLab: https://salsa.debian.org/debian-gis-team/pyshp/-/commit/9690919b08b52a477c40422b1e1a0d9f321e8203
--
View it on GitLab: https://salsa.debian.org/debian-gis-team/pyshp/-/commit/9690919b08b52a477c40422b1e1a0d9f321e8203
You're receiving this email because of your account on salsa.debian.org. Manage all notifications: https://salsa.debian.org/-/profile/notifications | Help: https://salsa.debian.org/help
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://alioth-lists.debian.net/pipermail/pkg-grass-devel/attachments/20260630/561ee1e4/attachment-0001.htm>
More information about the Pkg-grass-devel
mailing list